path: root/packages/meshbay-node/tests/test_security_regressions.py

"""
Phase 11.5 security regression tests.

Each test here encodes a finding from `second-review.md`. They are negative tests:
they assert that an attack does NOT work. The pre-11.5 code passed 209 feature
tests while every one of these attacks succeeded — the suite only ever exercised
happy paths, never an authorization boundary.

If one of these starts failing, a fix has been reverted. Do not "fix" the test.
"""

import base64
import struct
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from conftest import one_root
from meshbay_node.transport.webrtc_server import WebRTCPeerSession


def _safe_name_re():
    """
    Imported lazily so that a missing allowlist fails the two tests that need it,
    rather than aborting collection of the whole module and hiding every other
    finding's result.
    """
    from meshbay_node.transport.webrtc_server import SAFE_UPLOAD_NAME
    return SAFE_UPLOAD_NAME


# ── C1: the unauthenticated HTTP file API must stay deleted ───────────────────

def test_http_file_api_is_gone():
    """
    C1: transport/http_server.py served GET /index and GET /file/{id} on 0.0.0.0
    with no authentication, for private groups too. It was deleted rather than
    patched. Re-adding any module that serves file bytes outside the MNP handshake
    reintroduces a full confidentiality bypass.
    """
    with pytest.raises(ImportError):
        import meshbay_node.transport.http_server  # noqa: F401

    import meshbay_node.transport as transport
    assert not hasattr(transport, "create_http_app")


def test_tcp_transport_is_gone():
    """C6: the TCP+TLS server accepted a bare JWT with no GEK proof."""
    with pytest.raises(ImportError):
        import meshbay_node.transport.server  # noqa: F401

    import meshbay_node.transport as transport
    assert not hasattr(transport, "ChunkServer")


def test_daemon_exposes_no_plaintext_listener():
    """
    C1: the daemon must not bind anything that serves content without a handshake.
    NodeConfig no longer carries an HTTP port at all.
    """
    from meshbay_node.config import NodeConfig, GroupConfig

    assert "http_port" not in NodeConfig.__dataclass_fields__
    assert "http_port" not in GroupConfig.__dataclass_fields__
    assert "port" not in NodeConfig.__dataclass_fields__


# ── C5a: upload filename allowlist ───────────────────────────────────────────

@pytest.mark.parametrize("name", [
    "../../etc/passwd",
    "..\\windows\\system32",
    "/absolute/path",
    "<img src=x onerror=alert(1)>",          # the H2 stored-XSS vector
    'name";DROP TABLE x;--',
    ".hidden",
    "",
    "a" * 200,
    "file\x00.mp4",
    "sub/dir/file.mp4",
])
def test_upload_rejects_unsafe_filenames(name):
    """C5a/H2: only a conservative allowlist may reach the filesystem."""
    assert not _safe_name_re().match(name), f"should be rejected: {name!r}"


@pytest.mark.parametrize("name", [
    "movie.mp4",
    "My Holiday Video.mkv",
    "report-2026.pdf",
    "track_01.flac",
    # Reported 2026-08-16: an upload refused as "Invalid filename". The rule was
    # ASCII-only, so most of the world could not send a file, and — worse — it
    # rejected the "name (1).ext" form that _free_name produces itself, so the
    # node refused names it had chosen.
    "été.txt",
    "naïve café.jpg",
    "Ich möchte.pdf",
    "日本語.mp4",
    "rapport (1).pdf",
])
def test_upload_accepts_ordinary_filenames(name):
    """The allowlist must not break normal use, in any script."""
    assert _safe_name_re().match(name), f"should be accepted: {name!r}"


@pytest.mark.parametrize("name", [
    "trailing space ",
    "ends.with.dot.",
    "..",
    "a\u202eexe.txt",          # right-to-left override: hides the real extension
])
def test_upload_rejects_names_that_lie_about_themselves(name):
    """Widening to Unicode must not admit names that misrepresent the file."""
    assert not _safe_name_re().match(name), f"should be rejected: {name!r}"


def test_the_node_never_generates_a_name_it_would_refuse(tmp_path):
    """_free_name resolves a collision by appending " (n)"; that has to be legal."""
    from meshbay_node.transport.webrtc_server import _free_name
    (tmp_path / "clip.mp4").touch()
    (tmp_path / "clip (1).mp4").touch()
    chosen = _free_name(tmp_path, "clip.mp4")
    assert chosen not in ("clip.mp4", "clip (1).mp4")
    assert _safe_name_re().match(chosen), (
        f"the node picked {chosen!r} and would then reject it on the next upload")


def _uploads_dir(session) -> Path:
    """
    Where this session's uploads land: uploads/ inside the group's upload root.

    Asked of the root set rather than assembled by hand, so a test cannot pass
    while agreeing with a wrong answer the code also produced.
    """
    root = session._ctx["roots"].upload_root
    assert root is not None, "the fixture must designate an upload root"
    return root.path / "uploads"


def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession:
    """A peer session wired to a real shared root, with sending stubbed out."""
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node}

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = None
    session._user_id = user_id
    session._pk_user = ""
    session._uploads = {}
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def test_upload_cannot_overwrite_another_members_file(tmp_path):
    """
    C5a: uploads used to land in the shared root under a client-chosen name and
    overwrite whatever was there. That let any member destroy the operator's
    files, and — by becoming the recorded uploader of the replaced file — delete
    them through the uploader path, bypassing the Ed25519 admin challenge.

    The per-user quarantine that fixed it was removed on 2026-08-14: files now go
    where the member is looking, because a shared directory nobody can organise is
    not a shared directory. What made the quarantine work is kept, and is what
    this test now asserts — an existing file is never replaced.
    """
    victim = _session(tmp_path, "victim-user")
    uploads = _uploads_dir(victim)
    uploads.mkdir()
    original = uploads / "important.mp4"
    original.write_bytes(b"operator's original content")

    attacker = _session(tmp_path, "attacker-user")
    attacker._do_file_upload({
        "filename": "important.mp4",
        "chunk_index": 0,
        "total_chunks": 1,
        "data": base64.b64encode(b"attacker content").decode(),
    })

    assert original.read_bytes() == b"operator's original content", (
        "an upload replaced an existing file (C5a)")
    assert (uploads / "important (2).mp4").read_bytes() == b"attacker content"


def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
    """C5a: even the original uploader does not get to overwrite."""
    session = _session(tmp_path, "user-1")
    payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1,
               "data": base64.b64encode(b"first").decode()}
    session._do_file_upload(dict(payload))
    session.sent.clear()

    session._do_file_upload(dict(payload))
    uploads = _uploads_dir(session)
    assert (uploads / "movie.mp4").read_bytes() == b"first", (
        "the first upload was replaced")
    assert (uploads / "movie (2).mp4").read_bytes() == b"first"


@pytest.mark.parametrize("bad", [
    {"dir": "..", "name": "evil"},
    {"dir": "", "name": ".."},
    {"dir": "", "name": "a/b"},
    {"dir": "/etc", "name": "evil"},
    {"dir": "", "name": ".hidden"},
])
def test_dir_create_cannot_escape_the_shared_root(tmp_path, bad):
    """Creating a directory is not privileged, but it still writes to a disk."""
    session = _session(tmp_path, "user-1")
    before = set(tmp_path.rglob("*"))

    session._do_dir_create(bad)

    assert any(m.get("type") == "error" for m in session.sent), bad
    assert set(tmp_path.rglob("*")) == before, f"created something via {bad!r}"


def test_upload_ignores_any_directory_the_client_asks_for(tmp_path):
    """
    Uploads land in uploads/, chosen by the node. A client that names somewhere
    else — or nowhere at all — changes nothing, so the traversal surface that a
    client-chosen destination would open does not exist on this path.
    """
    session = _session(tmp_path, "user-1")

    session._do_file_upload({
        "filename": "note.txt", "dir": "../../etc",
        "chunk_index": 0, "total_chunks": 1,
        "data": base64.b64encode(b"x").decode(),
    })

    assert (_uploads_dir(session) / "note.txt").read_bytes() == b"x"
    assert not (tmp_path / "etc").exists()


def test_two_members_can_send_the_same_filename(tmp_path):
    """
    One shared uploads/ means collisions are ordinary — every camera produces
    IMG_1234.jpg. The second gets a free name; neither replaces the other.
    """
    first = _session(tmp_path, "user-1")
    first._do_file_upload({
        "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1,
        "data": base64.b64encode(b"first").decode(),
    })
    second = _session(tmp_path, "user-2")
    second._do_file_upload({
        "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1,
        "data": base64.b64encode(b"second").decode(),
    })

    uploads = _uploads_dir(first)
    assert (uploads / "IMG_1234.jpg").read_bytes() == b"first"
    assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second"

    ack = [m for m in second.sent if m.get("type") == "file_upload_ack"][-1]
    assert ack["stored_as"] == "IMG_1234 (2).jpg", (
        "the sender must be told the name that was used, or a chat attachment "
        "points at someone else's file")


# ── H1: group isolation ──────────────────────────────────────────────────────

def test_chat_store_and_peers_are_per_group(tmp_path):
    """
    H1: chat_store and the peer registry were read from the shared transport
    context, so on a multi-group node every group's messages went to the first
    group's database and were served back to members of every other group.
    """
    index_a = GroupIndex(group_id="a" * 32, sk_node=Ed25519PrivateKey.generate())
    index_b = GroupIndex(group_id="b" * 32, sk_node=Ed25519PrivateKey.generate())
    groups = {
        "a" * 32: {"chat_store": "STORE_A", "index": index_a, "roots": one_root(tmp_path / "a")},
        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "roots": one_root(tmp_path / "b")},
    }
    ctx = {"groups": groups}

    sess_a = WebRTCPeerSession.__new__(WebRTCPeerSession)
    sess_a._ctx, sess_a._group_id, sess_a._user_id = ctx, "a" * 32, "alice"

    sess_b = WebRTCPeerSession.__new__(WebRTCPeerSession)
    sess_b._ctx, sess_b._group_id, sess_b._user_id = ctx, "b" * 32, "bob"

    assert sess_a._group_ctx()["chat_store"] == "STORE_A"
    assert sess_b._group_ctx()["chat_store"] == "STORE_B"

    sess_a._peer_registry()["alice"] = sess_a
    sess_b._peer_registry()["bob"] = sess_b

    # Alice's broadcast target set must not contain Bob, who is in another group.
    assert "bob" not in sess_a._peer_registry()
    assert "alice" not in sess_b._peer_registry()

    sess_a._user_names()["alice"] = "Alice"
    assert "alice" not in sess_b._user_names()


def test_daemon_sets_no_global_chat_store(tmp_path):
    """H1: the daemon must not hoist one group's chat store onto the transport."""
    source = (Path(__file__).parent.parent
              / "src" / "meshbay_node" / "daemon.py").read_text(encoding="utf-8")
    assert '_ctx["chat_store"]' not in source, (
        "daemon must not assign a transport-wide chat_store — it leaks chat "
        "across groups (H1)"
    )


# ── H2: node admin UI escaping ───────────────────────────────────────────────

def test_no_member_can_hand_the_node_key_material(tmp_path):
    """
    C5b, strengthened by the invite redesign (docs/invite-pairing-v1.md).

    This test used to assert that `gek_bundle_store` answered with an admin
    challenge and stored nothing without an operator signature. The message is now
    gone entirely: the node holds the GEK and wraps it itself, so no member ever
    submits key material, authorized or not. Deleting the path is a stronger
    guarantee than gating it, which is why the assertion changed rather than the
    behaviour regressing.
    """
    from meshbay_common.protocol import MNP as _MNP

    assert not hasattr(_MNP, "GEK_BUNDLE_STORE"), (
        "the member-supplied bundle message is back — the node must never accept "
        "key material over MNP (C5b)"
    )

    source = (Path(__file__).parent.parent
              / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
    assert "_do_gek_bundle_store" not in source
    assert "_admin_exec_bundle_store" not in source


def test_unknown_message_stores_nothing(tmp_path):
    """A peer sending the retired message must not reach any storage path."""
    session = _session(tmp_path, "ordinary-member")
    session._group_id = None
    session._admin_ops = {}

    stored = []

    class _Store:
        async def store(self, *args):
            stored.append(args)

    session._ctx["bundle_store"] = _Store()
    session._handle_message({
        "type": "gek_bundle_store",
        "user_id": "victim", "group_id": "g" * 32,
        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
    })

    assert stored == [], "a retired message type still reached the bundle store"


def test_gek_auto_activation_is_gone():
    """
    C5b: the node used to unwrap and adopt any bundle addressed to the operator.
    Since the operator's X25519 public key is public, any member could hand the
    node a GEK of their choosing. Nothing arriving over MNP may set a live GEK.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
    assert "_try_activate_gek" not in source
    assert 'unwrap_gek_aes' not in source, (
        "the MNP path must not unwrap a GEK — activation is local-admin only"
    )


# ── H5: admin challenge is bound, not a blind signing oracle ─────────────────

def _transcript(**kw):
    from meshbay_common.adminop import admin_transcript
    base = dict(op="file_delete", node_pk_b64="NODEPK", group_id="g" * 32,
                subject="file-1", nonce=b"\x01" * 32, ts=1_700_000_000)
    base.update(kw)
    return admin_transcript(**base)


def test_admin_transcript_is_domain_separated():
    """H5: signatures here can never be valid in another MeshBay protocol."""
    assert _transcript().startswith(b"meshbay:admin:v1")


@pytest.mark.parametrize("field,value", [
    ("op", "invite_create"),
    ("subject", "file-2"),
    ("node_pk_b64", "OTHERNODE"),
    ("group_id", "h" * 32),
    ("nonce", b"\x02" * 32),
    ("ts", 1_700_000_001),
])
def test_admin_transcript_binds_every_field(field, value):
    """
    H5: a signature must not carry over to another operation, subject, node,
    group, challenge or moment in time.
    """
    assert _transcript() != _transcript(**{field: value}), (
        f"transcript ignores {field} — signature would be reusable"
    )


def test_admin_transcript_is_unambiguous():
    """
    H5/L4: fields are length-prefixed. With plain concatenation a crafted subject
    could impersonate the following field and two different operations would
    produce identical signed bytes.
    """
    a = _transcript(subject="file-1", group_id="g")
    b = _transcript(subject="1", group_id="gfile-")
    assert a != b, "concatenation is ambiguous — length prefixes missing"


def test_admin_signature_does_not_transfer_between_operations(tmp_path):
    """
    H5: the concrete attack. A signature collected to delete a file must not
    authorize storing a GEK bundle.
    """
    from meshbay_common.adminop import OP_FILE_DELETE, OP_INVITE_CREATE

    sk_admin = Ed25519PrivateKey.generate()
    delete_transcript = _transcript(op=OP_FILE_DELETE)
    signature = sk_admin.sign(delete_transcript)

    invite_transcript = _transcript(op=OP_INVITE_CREATE)
    with pytest.raises(Exception):
        sk_admin.public_key().verify(signature, invite_transcript)


def test_admin_challenge_expires(tmp_path):
    """H5: a stale challenge must not be usable."""
    import time as _time
    from meshbay_common.adminop import ADMIN_CHALLENGE_TTL, OP_FILE_DELETE

    session = _session(tmp_path, "operator")
    session._group_id = None
    session._admin_ops = {
        "op-1": {
            "op": OP_FILE_DELETE, "subject": "file-1", "nonce": b"\x00" * 32,
            "ts": int(_time.time()) - ADMIN_CHALLENGE_TTL - 5, "payload": {},
        }
    }
    session._do_admin_response({"op_id": "op-1", "signature": ""})
    assert any(m.get("type") == "error" and "expired" in m.get("detail", "").lower()
               for m in session.sent)


def test_denylist_persists_and_honours_groups(tmp_path):
    """
    H4: revocations lived only in memory, so a node restart silently un-revoked
    everyone, and 'group' targets were dropped entirely — the hub signed and
    broadcast them, the node's handler understood only 'user' and 'jti'.
    """
    from meshbay_node.transport import Denylist

    path = tmp_path / "denylist.json"
    first = Denylist(path=path)
    first.deny_group("g-revoked")
    first.deny_user("u-revoked")
    first.deny_jti("j-revoked")

    # A fresh instance stands in for a daemon restart.
    reloaded = Denylist(path=path)
    assert reloaded.is_denied("", "", "g-revoked"), "group revocation not honoured"
    assert reloaded.is_denied("u-revoked", "")
    assert reloaded.is_denied("", "j-revoked")
    assert not reloaded.is_denied("someone", "other", "g-allowed")


def test_swarm_registration_skips_private_groups():
    """
    H7: the daemon registered content hashes for every group, private included,
    handing the hub a fingerprint of every private file. The bug was masked by a
    mis-mounted route, so fixing the route without this filter would have turned a
    dormant leak into a live one.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "daemon.py").read_text(encoding="utf-8")
    assert 'visibility' in source and '_register_swarm' in source
    # Both registration sites must gate on public visibility.
    for marker in ['gctx.get("visibility") == "public"',
                   'group_cfg.visibility == "public"']:
        assert marker in source, f"swarm registration not gated: {marker}"


def test_keystore_argon2_is_production_strength():
    """M2: the keystore protects the node's private keys and sat at 64 MB."""
    from meshbay_common.crypto import ARGON2_MEMORY_COST
    assert ARGON2_MEMORY_COST >= 262144


def test_keystore_records_argon2_params_for_migration(tmp_path):
    """
    M2: raising the parameters must not orphan existing keystores, so each
    envelope records the parameters it was written with.
    """
    import json
    from meshbay_node.keystore import create_keystore, load_keystore

    path = tmp_path / "keystore.enc"
    created = create_keystore(path=path, password="correct horse battery")
    envelope = json.loads(path.read_text(encoding="utf-8"))
    assert envelope["argon2"]["memory_cost"] >= 262144

    reopened = load_keystore(path=path, password="correct horse battery")
    assert reopened.pk_ed25519_b64 == created.pk_ed25519_b64


def test_legacy_keystore_still_opens(tmp_path):
    """M2: a keystore written under the 64 MB profile must still unlock."""
    import base64 as _b64
    import json
    import msgpack
    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
    from meshbay_common.crypto import (
        LEGACY_ARGON2_ITERATIONS, LEGACY_ARGON2_LANES, LEGACY_ARGON2_MEMORY_COST,
        derive_keystore_key, encrypt_keystore, pk_to_b64, sk_to_b64,
    )
    from meshbay_node.keystore import load_keystore

    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
    payload = msgpack.packb({
        "sk_ed25519_b64": sk_to_b64(sk_ed),
        "sk_x25519_b64": sk_to_b64(sk_x),
    }, use_bin_type=True)

    salt = b"\x01" * 16
    key = derive_keystore_key(
        "legacy-pass", salt,
        iterations=LEGACY_ARGON2_ITERATIONS,
        memory_cost=LEGACY_ARGON2_MEMORY_COST,
        lanes=LEGACY_ARGON2_LANES,
    )
    iv, ct, tag = encrypt_keystore(payload, key)

    path = tmp_path / "legacy.enc"
    # No "argon2" key — exactly how pre-M2 envelopes look.
    path.write_text(json.dumps({
        "version": 1,
        "argon2_salt_b64": _b64.b64encode(salt).decode(),
        "iv_b64": _b64.b64encode(iv).decode(),
        "tag_b64": _b64.b64encode(tag).decode(),
        "ciphertext_b64": _b64.b64encode(ct).decode(),
    }))

    keys = load_keystore(path=path, password="legacy-pass")
    assert keys.pk_ed25519_b64 == pk_to_b64(sk_ed.public_key())
    assert keys.pk_x25519_b64 == pk_to_b64(sk_x.public_key())


def test_dead_gek_protocol_constants_removed():
    """L1: the node never serves a GEK; the message types should not suggest it."""
    from meshbay_common.protocol import MNP
    assert not hasattr(MNP, "GEK_REQUEST")
    assert not hasattr(MNP, "GEK_RESPONSE")


def test_peer_errors_do_not_leak_internals():
    """
    L3: arbitrary exception text carries filesystem paths and internal state, so
    the catch-all handler must not relay it.

    Deliberately narrow: HandshakeError messages ARE sent to the peer, because a
    client needs to know why it was refused, and those strings are authored for
    that purpose. The check targets the generic `except Exception as e` path.
    """
    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
    assert '"detail": str(e)' not in source, (
        "generic exception text relayed to peer — use a fixed message"
    )
    # And the catch-all must still exist, sending something opaque.
    assert '"detail": "Request failed"' in source


def test_pre_handshake_message_budget_is_small():
    """
    H6: the frame limit was a flat 64 MB applied before authentication, so an
    unauthenticated peer could announce a huge frame and dribble bytes into it.
    """
    from meshbay_node.transport.webrtc_server import (
        MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer,
    )
    assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024
    assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG

    buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
    buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x")
    with pytest.raises(ValueError):
        list(buf.messages())


def test_stream_segment_is_not_synchronous():
    """
    H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop,
    stalling every peer on the node for up to thirty seconds per request.

    Asserts the property (the worker is a coroutine, ffmpeg is spawned through
    asyncio) rather than grepping for "subprocess.run" — which also matches the
    comment that documents the old behaviour.
    """
    import ast
    import inspect
    from meshbay_node.transport.webrtc_server import WebRTCPeerSession

    assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async)

    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
              / "transport" / "webrtc_server.py").read_text(encoding="utf-8")
    tree = ast.parse(source)
    blocking = [
        node for node in ast.walk(tree)
        if isinstance(node, ast.Call)
        and isinstance(node.func, ast.Attribute)
        and node.func.attr == "run"
        and isinstance(node.func.value, ast.Name)
        and node.func.value.id == "subprocess"
    ]
    assert not blocking, "blocking subprocess.run() in the event loop"
    assert "_transcode_sem" in source, "ffmpeg spawns must be capped"


def test_pre_proof_fetches_are_bounded():
    """C4: the pre-proof bundle window is a disclosure surface; bound it."""
    from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES
    assert 0 < MAX_PRE_PROOF_FETCHES <= 10


def test_node_admin_ui_requires_token():
    """
    11.5.3: "localhost only" is not authentication. Any local process — or a
    rebound browser page — could re-initialise a group's GEK and read the audit log.
    """
    from fastapi.testclient import TestClient
    from meshbay_node.ui.app import create_ui_app

    app = create_ui_app({"status": "running", "groups_ctx": {},
                         "indexes": {}, "ui_token": "secret-token"})
    client = TestClient(app)

    assert client.get("/api/status").status_code == 403
    assert client.get("/api/status?t=wrong").status_code == 403
    assert client.get("/api/groups?t=wrong").status_code == 403
    assert client.get("/api/status?t=secret-token").status_code == 200
    assert client.get(
        "/api/status", headers={"X-MeshBay-Token": "secret-token"}
    ).status_code == 200


def test_node_control_api_serves_no_html():
    """
    H2 was stored XSS in the server-rendered admin dashboard: a member-chosen
    filename, or a hub-supplied username, landed in an HTML page on the
    operator's machine unescaped. That dashboard is gone
    (docs/refactor-node-ui.md phase 5) — the control API is JSON only, so there
    is no server-side template to inject into. The Node page that replaced it
    ships in the desktop client and escapes by default (Preact).

    This locks the removal in: the HTML routes stay 404, and the render helpers
    stay deleted so nothing reintroduces a template by importing one.
    """
    import meshbay_node.ui.app as ui_app
    from fastapi.testclient import TestClient

    app = ui_app.create_ui_app({"status": "running", "groups_ctx": {},
                                "indexes": {}, "ui_token": "t"})
    client = TestClient(app)
    for path in ("/", "/audit", "/dashboard"):
        assert client.get(f"{path}?t=t").status_code == 404, path

    for gone in ("_render_page", "_render_audit_page", "_render_roster",
                 "_AUDIT_HTML"):
        assert not hasattr(ui_app, gone), f"{gone} came back — HTML surface"