fix: stop a stream on close, count only real users, record where a node is

**Closing the viewer left the node working.** Nothing told it to stop:
the player dropped its handlers, which only made the browser deaf. ffmpeg
kept running and held one of the node's two transcode slots until the
credit timeout expired two minutes later — which is why the next video
answered "server busy". `stream_stop` ends it at once, and the viewer
also drops its queue, ends the MediaSource and revokes the object URL on
the way out, any of which could be holding megabytes of decrypted video.

While there: `file_chunk` replies were matched to their requests by
arrival order, which was true by luck rather than by construction. The
reply now names the file it belongs to and is matched on that and the
chunk index; a chunk nobody is waiting for is dropped instead of being
handed to whatever request happens to be oldest.

**The administration panel counted its own history.** A deleted account
is tombstoned so the connection log stays readable, and every count and
list treated that row as a user — including a group's member count, and
the member list of the group itself. They do not any more.

**Where a node is.** `endpoint_hint` is what a node believes its address
to be, learned from a STUN server and sent to us: useful for reaching it,
and a claim. The announcement that carries it is signed with the node key
over a fresh timestamp, so the address that request *arrives from* is the
address of whoever holds that key — that is now recorded on the node row
and shown in a Nodes tab, next to the hint, with the difference spelled
out. Clients get the same treatment: `webrtc_offer` is logged with the
address the hub saw when a browser starts a peer connection.

Verified against the live deployment: the node's row reads 90.112.206.172
after a restart, and in e2e a stopped stream goes quiet in one message
and the next one starts immediately instead of being refused.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

1 files changed, 150 insertions, 0 deletions

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"