diff options
Diffstat (limited to 'packages/meshbay-hub')
11 files changed, 383 insertions, 11 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/admin.py b/packages/meshbay-hub/src/meshbay_hub/api/admin.py index 785731d..7ee05ca 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/admin.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/admin.py @@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub.auth import decrypt_email from meshbay_hub.api.deps import require_admin, require_moderator -from meshbay_hub.api.revocation import get_connected_node_count +from meshbay_hub.api.revocation import get_connected_node_count, is_node_connected from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import Group, GroupMember, IPLog, Node, User @@ -42,8 +42,18 @@ async def admin_stats( current_user: User = Depends(require_moderator), db: AsyncSession = Depends(get_db), ): - user_count = (await db.execute(select(func.count()).select_from(User))).scalar_one() - group_count = (await db.execute(select(func.count()).select_from(Group))).scalar_one() + # Deleted accounts are tombstoned rather than dropped, so that the + # connection log stays readable. They are not users any more and must not be + # counted as any: a hub whose user count only ever rises is measuring its + # own history, not its population. + user_count = (await db.execute( + select(func.count()).select_from(User) + .where(User.status != "deleted"))).scalar_one() + # Groups are not tombstoned — deleting one removes the row — so every group + # here is a group. A revoked one is suspended by moderation and still shown + # in the list, so counting it keeps the two consistent. + group_count = (await db.execute( + select(func.count()).select_from(Group))).scalar_one() node_count = (await db.execute(select(func.count()).select_from(Node))).scalar_one() return { "users": user_count, @@ -63,14 +73,16 @@ async def admin_list_users( offset: int = 0, limit: int = Query(default=50, le=200), ): - query = select(User).order_by(User.created_at.desc()) + query = (select(User).where(User.status != "deleted") + .order_by(User.created_at.desc())) if q: query = query.where(User.username.ilike(f"%{q}%")) query = query.offset(offset).limit(limit) result = await db.execute(query) users = result.scalars().all() - total_query = select(func.count()).select_from(User) + total_query = (select(func.count()).select_from(User) + .where(User.status != "deleted")) if q: total_query = total_query.where(User.username.ilike(f"%{q}%")) total = (await db.execute(total_query)).scalar_one() @@ -222,9 +234,13 @@ async def admin_list_groups( query = ( select( Group, - func.count(GroupMember.user_id).label("member_count"), + func.count(User.id).label("member_count"), ) + # Members, not rows: a deleted account's membership is removed with it, + # but joining through User keeps the count honest if one ever survives. .outerjoin(GroupMember, Group.id == GroupMember.group_id) + .outerjoin(User, (User.id == GroupMember.user_id) + & (User.status != "deleted")) .group_by(Group.id) .order_by(Group.created_at.desc()) .offset(offset) @@ -288,6 +304,44 @@ async def admin_patch_group( # ── IP Audit Logs ──────────────────────────────────────────────────────────── +@router.get("/nodes") +async def admin_list_nodes( + current_user: User = Depends(require_moderator), + db: AsyncSession = Depends(get_db), + limit: int = Query(default=100, le=200), +): + """ + Registered nodes, with the address the hub saw them announce from. + + `observed_ip` is the one to answer a question with: it comes from the + connection that carried a valid Ed25519 signature over a fresh timestamp, so + it is the address of whoever holds the node key. `endpoint_hint` is what the + node believes its own address to be, discovered through a STUN server and + sent to us — useful for reaching it, and not evidence of anything. + """ + rows = (await db.execute( + select(Node, User.username) + .outerjoin(User, User.id == Node.user_id) + .order_by(Node.announced_at.desc()) + .limit(limit))).all() + return { + "nodes": [ + { + "id": n.id, + "user_id": n.user_id, + "username": uname or "", + "pk_node": n.pk_node, + "observed_ip": n.observed_ip or "", + "endpoint_hint": n.endpoint_hint or "", + "last_seen": n.last_seen.isoformat() if n.last_seen else "", + "announced_at": n.announced_at.isoformat(), + "online": is_node_connected(n.id), + } + for n, uname in rows + ], + } + + @router.get("/logs") async def admin_list_logs( current_user: User = Depends(require_moderator), diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py index 74c6c9e..8e3197c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py @@ -207,7 +207,7 @@ async def group_members( result = await db.execute( select(User.id, User.username) .join(GroupMember, User.id == GroupMember.user_id) - .where(GroupMember.group_id == group_id) + .where(GroupMember.group_id == group_id, User.status != "deleted") ) members = [{"user_id": uid, "username": uname} for uid, uname in result.all()] return { diff --git a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py index 0770148..67e65f2 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py @@ -1,5 +1,6 @@ """Node endpoints — /v1/nodes/*""" +from datetime import datetime, timezone import base64 import time @@ -130,10 +131,18 @@ async def announce_node( select(Node).where(Node.user_id == current_user.id, Node.pk_node == body.pk_node)) node = existing.scalar_one_or_none() + # The address is taken from the connection, never from the body: the + # signature above proves who is announcing, and this is where they are + # announcing from. What the node believes its address to be — endpoint_hint, + # learned from a STUN server — is kept separately and is not evidence. + seen_from = client_ip(request) + if node is not None: node.endpoint_hint = body.endpoint_hint + node.observed_ip = seen_from + node.last_seen = datetime.now(timezone.utc) db.add(IPLog(user_id=current_user.id, event="node_announce", - ip_address=client_ip(request), detail=body.endpoint_hint)) + ip_address=seen_from, detail=body.endpoint_hint)) await db.commit() return {"node_id": node.id} @@ -141,12 +150,14 @@ async def announce_node( user_id=current_user.id, pk_node=body.pk_node, endpoint_hint=body.endpoint_hint, + observed_ip=seen_from, + last_seen=datetime.now(timezone.utc), ) db.add(node) db.add(IPLog( user_id=current_user.id, event="node_announce", - ip_address=client_ip(request), + ip_address=seen_from, detail=body.endpoint_hint, )) await db.commit() diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index 58ebf50..1f6d7f0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -55,6 +55,10 @@ _node_groups: dict[str, list[str]] = {} # node_id → [group_id, ...] _punch_events: dict[str, asyncio.Event] = {} # node_id → signaling event +def is_node_connected(node_id: str) -> bool: + return node_id in _connected_nodes + + def get_connected_node_count() -> int: return len(_connected_nodes) diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py index 8f84163..cb00a67 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py @@ -25,7 +25,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub.api.deps import get_current_user from meshbay_hub.api.middleware import limiter from meshbay_hub.db.engine import get_db -from meshbay_hub.db.models import Group, GroupMember, User +from meshbay_hub.api.netutil import client_ip +from meshbay_hub.db.models import Group, GroupMember, IPLog, User log = logging.getLogger(__name__) @@ -77,6 +78,14 @@ async def webrtc_offer( if len(body.sdp) > MAX_SDP_BYTES: raise HTTPException(status_code=413, detail="SDP too large") + # Logged here because this is the moment a browser starts a peer connection, + # and the address it starts it from is this one — the hub's own view of the + # TCP connection. Whatever address the peers then discover through STUN is + # theirs to negotiate and is not what a log should record. + db.add(IPLog(user_id=current_user.id, event="webrtc_offer", + ip_address=client_ip(request), detail=node_id[:8])) + await db.commit() + ws = _connected_nodes.get(node_id) if not ws: raise HTTPException(status_code=404, detail="Node not connected") diff --git a/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py new file mode 100644 index 0000000..c720cab --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py @@ -0,0 +1,35 @@ +"""node_observed_ip + +Where a node was actually seen, as opposed to where it says it is. + +`endpoint_hint` is discovered by the node through a STUN server and sent to the +hub in its announcement — useful for reaching it, and a claim. `observed_ip` is +the address the announcement itself arrived from, on a request carrying a valid +Ed25519 signature over a fresh timestamp, so it is the address of whoever holds +the node key. That is the one an administrator's question is about. + +Revision ID: d1f47a90c3b2 +Revises: c5e93b1a2f60 +Create Date: 2026-08-15 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = 'd1f47a90c3b2' +down_revision: Union[str, Sequence[str], None] = 'c5e93b1a2f60' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column('nodes', sa.Column('observed_ip', sa.String(45), nullable=True)) + op.add_column('nodes', + sa.Column('last_seen', sa.DateTime(timezone=True), nullable=True)) + + +def downgrade() -> None: + op.drop_column('nodes', 'last_seen') + op.drop_column('nodes', 'observed_ip') diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py index 2e36989..7d5a3e3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/db/models.py +++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py @@ -72,6 +72,13 @@ class Node(Base): user_id: Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False) pk_node: Mapped[str] = mapped_column(String(64), nullable=False) # Ed25519 b64 endpoint_hint: Mapped[str | None] = mapped_column(String(128)) # "ip:port" or null + # The address the hub saw this node connect from, recorded on an announce + # that carried a valid Ed25519 signature over a fresh timestamp. Unlike + # endpoint_hint — which the node discovers through STUN and sends us — this + # is observed rather than claimed, and it is the one to answer questions + # with. IPv4 or IPv6. + observed_ip: Mapped[str | None] = mapped_column(String(45)) + last_seen: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) announced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) user: Mapped["User"] = relationship(back_populates="nodes") diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js index 18ce83a..cf33d72 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/app.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js @@ -2587,10 +2587,25 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) { videoRef.current.removeEventListener('seeking', onSeeking); } if (transport) { + // Tell the node first: dropping the handlers only makes us deaf, and a + // stream nobody is listening to still occupies a transcode slot. + transport.stopStream(); transport.onStreamInit = null; transport.onStreamData = null; transport.onStreamEnd = null; } + // The queue can hold several megabytes of decrypted video. + queueRef.current = []; + const ms = msRef.current; + if (ms && ms.readyState === 'open') { + try { ms.endOfStream(); } catch { /* already ended */ } + } + if (blobUrlRef.current) { + URL.revokeObjectURL(blobUrlRef.current); + blobUrlRef.current = null; + } + sbRef.current = null; + msRef.current = null; }; }, [entry, flushQueue]); @@ -3056,6 +3071,7 @@ function AdminPage({ token }) { const [logEvent, setLogEvent] = useState(''); const [logOffset, setLogOffset] = useState(0); const [blocklist, setBlocklist] = useState([]); + const [nodes, setNodes] = useState([]); const [detailUser, setDetailUser] = useState(null); const [error, setError] = useState(''); @@ -3105,6 +3121,10 @@ function AdminPage({ token }) { if (tab === 'stats') loadStats(); else if (tab === 'users') loadUsers(userSearch); else if (tab === 'groups') loadGroups(); + else if (tab === 'nodes') { + hubFetch('/v1/admin/nodes', { token }) + .then(d => setNodes(d.nodes || [])).catch(e => setError(e.message)); + } else if (tab === 'logs') { setLogOffset(0); loadLogs(logEvent, 0); } else if (tab === 'blocklist') loadBlocklist(); }, [tab]); @@ -3157,7 +3177,7 @@ function AdminPage({ token }) { } catch (e) { setError(e.message); } }, [token]); - const TABS = ['stats', 'users', 'groups', 'logs', 'blocklist']; + const TABS = ['stats', 'users', 'groups', 'nodes', 'logs', 'blocklist']; return html` <div> @@ -3267,6 +3287,41 @@ function AdminPage({ token }) { </table> `} + ${tab === 'nodes' && html` + <p class="settings-hint" style="margin-bottom:10px"> + ${t('admin.nodes_hint')} + </p> + <table class="admin-table"> + <thead><tr> + <th>${t('admin.col_username')}</th> + <th>${t('admin.col_observed_ip')}</th> + <th>${t('admin.col_hint')}</th> + <th>${t('admin.col_last_seen')}</th> + <th>${t('admin.col_status')}</th> + </tr></thead> + <tbody> + ${nodes.length === 0 && html` + <tr><td colspan="5" class="admin-empty">${t('admin.no_nodes')}</td></tr> + `} + ${nodes.map(n => html` + <tr key=${n.id}> + <td>${n.username || n.user_id.slice(0, 8)}</td> + <td style="font-family:monospace">${n.observed_ip || '—'}</td> + <td style="font-family:monospace;color:var(--text-dim)"> + ${n.endpoint_hint || '—'} + </td> + <td>${n.last_seen ? new Date(n.last_seen).toLocaleString() : '—'}</td> + <td> + <span class="badge ${n.online ? 'badge-ok' : ''}"> + ${n.online ? t('admin.node_online') : t('admin.node_offline')} + </span> + </td> + </tr> + `)} + </tbody> + </table> + `} + ${tab === 'logs' && html` <div class="admin-toolbar"> <select class="admin-select" value=${logEvent} onChange=${e => { diff --git a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js index 71f6cce..affa7c8 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js @@ -195,7 +195,19 @@ const en = { 'admin.tab_stats': 'Stats', 'admin.tab_users': 'Users', 'admin.tab_groups': 'Groups', + 'admin.tab_nodes': 'Nodes', 'admin.tab_logs': 'Logs', + 'admin.col_observed_ip': 'Seen from', + 'admin.col_hint': 'Announced hint', + 'admin.col_last_seen': 'Last announce', + 'admin.node_online': 'online', + 'admin.node_offline': 'offline', + 'admin.no_nodes': 'No nodes registered', + 'admin.nodes_hint': '"Seen from" is the address the announcement arrived from, ' + + 'on a request signed with the node key — that is the one to answer a ' + + 'question with. The hint is what the node believes its own address to be, ' + + 'learned from a STUN server and sent to us; it is useful for reaching the ' + + 'node and is not evidence of anything.', 'admin.tab_blocklist': 'Blocklist', // Admin stats diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index 844a201..df2abb0 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -562,6 +562,18 @@ class MeshBayTransport { } /** + * Nobody is watching any more. + * + * Closing the viewer used to say nothing to the node, which went on + * transcoding and holding one of its two slots until the credit timeout — so + * the next video answered "server busy". + */ + stopStream() { + if (!this._connected) return; + try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ } + } + + /** * Push a whole file, several chunks in flight at once. * * One chunk per round trip is 48 KB of throughput per RTT no matter how much @@ -778,6 +790,11 @@ class MeshBayTransport { }, 30000); this._pending.set(id, { _reqType: obj.type, + // Chunks are the one request that runs several at a time and can be + // interleaved with anything else on the channel. Matching them by + // arrival order was only ever true by luck; this makes it true. + _key: obj.type === 'file_req' + ? `chunk:${obj.file_id}:${obj.chunk_index}` : null, resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); }, reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); }, }); @@ -857,6 +874,24 @@ class MeshBayTransport { return; } + if (msg.type === 'file_chunk') { + const key = `chunk:${msg.file_id}:${msg.chunk_index}`; + for (const [, handler] of this._pending) { + // A node from before the reply carried a file_id: fall back to the + // index, which is still better than the oldest pending request. + const match = msg.file_id + ? handler._key === key + : handler._key && handler._key.endsWith(`:${msg.chunk_index}`); + if (match) { + handler.resolve(msg); + return; + } + } + // Nobody asked for it any more — a cancelled download, most likely. It + // must not be handed to whatever request happens to be waiting. + return; + } + const oldest = this._pending.entries().next(); if (!oldest.done) { const [, handler] = oldest.value; diff --git a/packages/meshbay-hub/tests/test_admin_views.py b/packages/meshbay-hub/tests/test_admin_views.py new file mode 100644 index 0000000..29f41d8 --- /dev/null +++ b/packages/meshbay-hub/tests/test_admin_views.py @@ -0,0 +1,150 @@ +""" +What the administration panel counts and lists. + +Deleting an account leaves a tombstone so the connection log stays readable +(see test_account_deletion). That row is not a user any more, and every place +that counts or lists people has to agree — a hub whose user count only ever +rises is measuring its own history rather than its population. +""" + +import base64 +import hashlib + +import pytest +from sqlalchemy import select + +from meshbay_hub.db.models import User + + +def _auth_key(password: str, username: str) -> str: + salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest() + return base64.b64encode( + hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode() + + +async def _user(client, username, password="a-long-enough-passphrase"): + await client.post("/v1/users/register", json={ + "username": username, "email": f"{username}@example.com", + "auth_key": _auth_key(password, username)}) + r = await client.post("/v1/users/login", json={ + "username": username, "auth_key": _auth_key(password, username)}) + return {"Authorization": f"Bearer {r.json()['access_token']}"} + + +async def _admin(client, db_session, username="root"): + headers = await _user(client, username) + user = (await db_session.execute( + select(User).where(User.username == username))).scalar_one() + user.role = "admin" + await db_session.commit() + return headers + + +@pytest.mark.asyncio +async def test_a_deleted_account_stops_being_counted(client, db_session): + admin = await _admin(client, db_session) + leaver = await _user(client, "ghost") + + before = (await client.get("/v1/admin/stats", headers=admin)).json()["users"] + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "ghost")}) + after = (await client.get("/v1/admin/stats", headers=admin)).json()["users"] + + assert after == before - 1, "a tombstone is still being counted as a user" + + +@pytest.mark.asyncio +async def test_a_deleted_account_is_not_listed(client, db_session): + admin = await _admin(client, db_session, "root2") + leaver = await _user(client, "vanishing") + + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "vanishing")}) + + r = await client.get("/v1/admin/users", headers=admin) + names = [u["username"] for u in r.json()["users"]] + assert "vanishing" not in names + assert not any(n.startswith("deleted-") for n in names), \ + "the tombstone is showing under its placeholder name" + assert r.json()["total"] == len(names) + + +@pytest.mark.asyncio +async def test_the_member_list_of_a_group_skips_them(client, db_session): + admin = await _admin(client, db_session, "root3") + owner = await _user(client, "host3") + leaver = await _user(client, "quitter") + + g = await client.post("/v1/groups", json={"name": "party"}, headers=owner) + gid = g.json()["group_id"] + await client.post(f"/v1/groups/{gid}/members/quitter", json={}, headers=owner) + + await client.request("DELETE", "/v1/users/me", headers=leaver, + json={"auth_key": _auth_key( + "a-long-enough-passphrase", "quitter")}) + + members = (await client.get(f"/v1/groups/{gid}/members", + headers=owner)).json()["members"] + assert [m["username"] for m in members] == ["host3"] + + groups = (await client.get("/v1/admin/groups", headers=admin)).json()["groups"] + party = next(g for g in groups if g["name"] == "party") + assert party["member_count"] == 1, \ + f"the group still counts its departed member ({party['member_count']})" + + +# ── Where a node actually is ──────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_node_is_recorded_at_the_address_it_announced_from( + client, db_session): + """ + The hub sees the connection; the node only tells us what a STUN server told + it. The observed address is the evidence, and the announcement carries an + Ed25519 signature over a fresh timestamp, so it is the address of whoever + holds the node key. + """ + import base64 + import time + + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives import serialization + + from meshbay_hub.db.models import Node + + admin = await _admin(client, db_session, "root4") + owner = await _user(client, "nodeowner") + uid = (await db_session.execute( + select(User.id).where(User.username == "nodeowner"))).scalar_one() + + sk = Ed25519PrivateKey.generate() + pk_b64 = base64.b64encode(sk.public_key().public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw)).decode() + ts = int(time.time()) + message = f"meshbay:node_announce:{uid}:{pk_b64}:{ts}".encode() + + r = await client.post("/v1/nodes/announce", headers=owner, json={ + "pk_node": pk_b64, + # A claim, and a deliberately implausible one. + "endpoint_hint": "203.0.113.42:19001", + "timestamp": ts, + "signature": base64.b64encode(sk.sign(message)).decode(), + }) + assert r.status_code in (200, 201), r.text + + node = (await db_session.execute( + select(Node).where(Node.pk_node == pk_b64))).scalar_one() + assert node.observed_ip, "no address was recorded for the announcement" + assert node.observed_ip != "203.0.113.42", \ + "the hub recorded what the node claimed instead of what it saw" + assert node.last_seen is not None + + listed = (await client.get("/v1/admin/nodes", headers=admin)).json()["nodes"] + row = next(n for n in listed if n["pk_node"] == pk_b64) + assert row["observed_ip"] == node.observed_ip + assert row["endpoint_hint"] == "203.0.113.42:19001", \ + "the claim is still shown, just not as the answer" + assert row["username"] == "nodeowner" |