"""
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.crypto import generate_gek
from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from conftest import one_root, opened_ack, sealed_upload
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 an unaddressed upload lands: the first writable root itself.

    There is no `uploads/` subdirectory any more. It was the last of v5's
    quarantine — the per-user layer went on 2026-08-14 — and it went for the
    same reason: a folder appearing beside the operator's library because
    somebody sent a file is the node deciding how their disk is arranged. The
    protections that made the quarantine worth having are the allowlist, the
    size cap, the chunk ordering and the no-overwrite rule, and every one of
    them is asserted below, unchanged.

    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.
    """
    writable = session._ctx["roots"].writable_roots
    assert writable, "the fixture must give the group a writable root"
    return writable[0].path


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())
    # A group key, because uploads are sealed under it since MNP 2.0 — the
    # handler opens the payload before it has a filename to refuse.
    ctx = {"roots": one_root(shared_root), "index": index,
           "sk_node": index.sk_node, "gek": generate_gek()}

    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)
    original = uploads / "important.mp4"
    original.write_bytes(b"operator's original content")

    attacker = _session(tmp_path, "attacker-user")
    attacker._do_file_upload(sealed_upload(
        attacker, filename="important.mp4", data=b"attacker content"))

    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")
    def _send_it():
        # Sealed afresh each time: a nonce is drawn per message, so re-sending
        # the same dict would be a replay rather than a second upload.
        session._do_file_upload(sealed_upload(
            session, filename="movie.mp4", data=b"first"))

    _send_it()
    session.sent.clear()
    _send_it()
    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_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path):
    """
    The destination is now the folder the sender is looking at, which means the
    client does choose it — and the whole of what keeps that safe is that the
    choice is *resolved against the group's own roots* rather than joined to
    one.

    `RootSet.resolve()` refuses `..`, absolute segments and anything whose
    resolved form escapes its root, symlinks included. So "which of this
    group's folders" is answerable by a member and "which path on the
    operator's disk" is not.
    """
    session = _session(tmp_path, "user-1")
    (session._ctx["roots"].roots[0].path / "sub").mkdir()
    before = set(tmp_path.rglob("*"))

    for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc",
                "nope", "shared/missing"):
        session.sent.clear()
        session._do_file_upload(sealed_upload(
            session, filename="note.txt", data=b"x", dir=bad))
        refusal = [m for m in session.sent if m.get("type") == "error"]
        assert refusal, f"{bad!r} was accepted"
        assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad

    assert set(tmp_path.rglob("*")) == before, "a refused upload still wrote"


def test_an_upload_lands_in_the_folder_it_names(tmp_path):
    """
    And in that folder itself — the `uploads/` subdirectory the node used to
    create is gone. Somebody dropping a file into the folder they are looking
    at expects it to be in that folder.
    """
    session = _session(tmp_path, "user-1")
    root = session._ctx["roots"].roots[0]
    (root.path / "Albums").mkdir()

    session._do_file_upload(sealed_upload(
        session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums"))

    assert (root.path / "Albums" / "note.txt").read_bytes() == b"x"
    assert not (root.path / "Albums" / "uploads").exists(), (
        "the node invented a subdirectory in the operator's library")
    assert not (root.path / "uploads").exists()


def test_an_upload_goes_to_the_root_it_names(tmp_path):
    """
    With two writable roots there is no defensible default, and the client is
    the only party that knows which directory the person is looking at. The
    node picking one meant a file uploaded from a folder on screen landed in a
    different one — the same "uploads went somewhere else" the single upload
    root was never allowed to guess about.
    """
    media = tmp_path / "Media"
    incoming = tmp_path / "Incoming"
    media.mkdir()
    incoming.mkdir()
    session = _session(tmp_path, "user-1")
    session._ctx["roots"] = RootSet.build([
        {"path": str(media), "writable": True},
        {"path": str(incoming), "writable": True},
    ])

    session._do_file_upload(sealed_upload(
        session, filename="note.txt", data=b"x", dir="Incoming"))

    assert (incoming / "note.txt").read_bytes() == b"x"
    assert not (media / "note.txt").exists(), "it went to the first root instead"


def test_a_read_only_root_refuses_an_upload(tmp_path):
    """
    RO is the mechanism now, not a hidden button. It binds the operator too:
    "read-only for everyone" is what makes a published library one, and an
    exception for whoever happens to hold admin authority is the sort of
    carve-out that later reads as the rule.
    """
    published = tmp_path / "Published"
    published.mkdir()
    session = _session(tmp_path, "user-1")
    session._ctx["roots"] = RootSet.build([{"path": str(published)}])
    session._is_node_admin = lambda: True

    session._do_file_upload(sealed_upload(
        session, filename="note.txt", data=b"x", dir="Published"))

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_read_only"
    assert not (published / "note.txt").exists()


def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path):
    """
    An MNP 1.0 client names no root, so the node falls back to the first
    writable one. There isn't one here, and the fallback must refuse rather
    than write into whatever root happens to come first.
    """
    published = tmp_path / "Published"
    published.mkdir()
    session = _session(tmp_path, "user-1")
    session._ctx["roots"] = RootSet.build([{"path": str(published)}])

    session._do_file_upload(sealed_upload(
        session, filename="note.txt", data=b"x"))

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "no_writable_root"
    assert not (published / "note.txt").exists()


def test_an_ejected_root_refuses_an_upload(tmp_path):
    """
    Writing to a drive somebody has their hand on is the thing eject exists to
    stop. `writable` is still true — that is configuration — so availability
    has to be checked separately, which is what an earlier version conflated.
    """
    usb = tmp_path / "USB"
    usb.mkdir()
    session = _session(tmp_path, "user-1")
    roots = RootSet.build([{"path": str(usb), "writable": True,
                            "removable": True}])
    roots.roots[0].ejected = True
    roots.roots[0].available = False
    session._ctx["roots"] = roots

    session._do_file_upload(sealed_upload(
        session, filename="note.txt", data=b"x", dir="USB"))

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_unavailable"
    assert not (usb / "note.txt").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(sealed_upload(
        first, filename="IMG_1234.jpg", data=b"first"))
    second = _session(tmp_path, "user-2")
    # Same group, so the same key: `_session` builds one per call, and two
    # members of one group do not have two.
    second._ctx["gek"] = first._ctx["gek"]
    second._do_file_upload(sealed_upload(
        second, filename="IMG_1234.jpg", data=b"second"))

    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 "stored_as" not in ack, "the name the node chose must be sealed"
    assert opened_ack(second, 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_no_transport_ships_media_outside_the_aead():
    """
    `stream_seg` served an MPEG-TS segment as base64 with no encryption at all
    — the one content-plane message that never went through a GEK-derived key,
    on both transports, answering any authenticated member. Its browser caller
    was defined and never invoked. Removed in MNP 2.0 rather than repaired:
    `stream_data` already does the job under `chunk_ciphertext`.

    Asserted as the property, not as "the function is gone": what matters is
    that no transport has a field carrying media bytes past the AEAD. The old
    H6 test lived here — it pinned `_do_stream_segment_async` to a coroutine so
    ffmpeg could not block the event loop — and the handler outliving that
    concern is exactly what this replaces.
    """
    import re

    from meshbay_common.protocol import MNP

    assert not hasattr(MNP, "STREAM_SEGMENT"), (
        "the constant outliving the handlers is how a deleted endpoint keeps "
        "looking like part of the wire contract")

    root = Path(__file__).parent.parent / "src" / "meshbay_node" / "transport"
    for name in ("webrtc_server.py", "quic_server.py", "quic_client.py"):
        source = (root / name).read_text(encoding="utf-8")
        # Word boundaries: `_stream_segments` and `STREAM_SEGMENT_SIZE` belong
        # to the live `stream_data` path, which is encrypted and stays.
        assert not re.search(r"\bstream_seg\b", source), (
            f"{name} still speaks stream_seg")
        assert not re.search(r"\bSTREAM_SEGMENT\b", source), (
            f"{name} still names the removed type")
        assert "data_b64" not in source, (
            f"{name} carries a base64 media field — media leaves this node "
            "encrypted or not at all")


def test_ffmpeg_never_blocks_the_event_loop():
    """
    H6, the half that survives `stream_seg`: the live streaming path still
    spawns ffmpeg, and a synchronous spawn stalls every peer on the node.
    """
    import ast

    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"


# ── NS4 / M3: the node's own controls take no authority from the hub ─────────

async def _owner_session(tmp_path, roster, *, device: str = "",
                         confirmed: bool = False):
    """A session whose token says it is the account this node belongs to.

    Which is all a hub can decide: `_user_id` is the `sub` of a JWT it issued,
    so this is what an active hub forging a token arrives holding. Whether the
    *device* on the connection is one the node pinned as an operator is the
    other half, and no token can assert it.
    """
    from meshbay_node.indexer.group_index import GroupIndex

    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roots": one_root(shared), "index": index, "sk_node": index.sk_node,
        "gek": generate_gek(), "roster": roster,
        "node_user_id": "the-owner",
        "daemon_state": {"groups_ctx": {}, "reload_fn": None},
    }
    session._group_id = "g" * 32
    session._user_id = "the-owner"
    session._username = "the-owner"
    session._pinned_pk = device
    session._device_confirmed = confirmed
    session._pk_user = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


@pytest.mark.asyncio
async def test_a_token_naming_the_owner_is_not_node_authority(tmp_path):
    """
    The hub holds no user keys, so it cannot countersign a device — but it does
    choose what a token says. Six node-wide controls used to be gated on the
    account id alone, which is the hub's to decide: `node_status` (every group
    on the machine, with the operator's absolute paths), `node_settings_set`,
    `roster_read`, `denylist_read`, `denylist_clear` (the persisted revocation
    H4 exists to keep) and `node_reload`.

    An active hub reaches a completed handshake wherever it can also obtain the
    group key, which §3.5 concedes it can in an open-join group. From there,
    "the hub says you are the owner" was the whole of the check.
    """
    from meshbay_node.roster import Roster

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        forged = await _owner_session(tmp_path, roster)
        await forged._do_node_status({})
        assert forged.sent[-1]["type"] == "error"
        assert forged.sent[-1]["code"] == "not_operator"

        forged.sent.clear()
        await forged._do_denylist_read({})
        assert forged.sent[-1]["type"] == "error"

        forged.sent.clear()
        await forged._do_denylist_clear({"subject": ""})
        assert forged.sent[-1]["type"] == "error", (
            "clearing the denylist undoes a revocation every node enforces")
    finally:
        await roster.close()


@pytest.mark.asyncio
async def test_an_identified_device_that_is_not_an_operator_is_refused(tmp_path):
    """Being a pinned member of the group is not being the node's operator.

    The device proof is real here; what it proves is an ordinary member's key,
    and `operator_pks()` is rebuilt from the roster on every call so a revoked
    one stops working at once.
    """
    from meshbay_node.roster import Roster
    from meshbay_common.crypto import pk_to_b64

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        sk = Ed25519PrivateKey.generate()
        member_pk = pk_to_b64(sk.public_key())
        await roster.pin_identity("the-owner", "the-owner", member_pk,
                                  member_pk, "code")
        session = await _owner_session(tmp_path, roster, device=member_pk,
                                       confirmed=True)
        await session._do_node_status({})
        assert session.sent[-1]["type"] == "error"
    finally:
        await roster.close()


@pytest.mark.asyncio
async def test_a_paired_operator_device_is_what_opens_it(tmp_path):
    """The positive case, so the test above is about authority and not about
    everything being refused."""
    from meshbay_node.roster import Roster
    from meshbay_common.join import ROLE_OPERATOR
    from meshbay_common.crypto import pk_to_b64

    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        sk = Ed25519PrivateKey.generate()
        pk = pk_to_b64(sk.public_key())
        await roster.pin_identity("the-owner", "the-owner", pk, pk, "code")
        await roster.set_member("", "the-owner", ROLE_OPERATOR, "active",
                                "local-cli")
        session = await _owner_session(tmp_path, roster, device=pk,
                                       confirmed=True)
        await session._do_denylist_read({})
        assert session.sent[-1]["type"] != "error", session.sent[-1]
    finally:
        await roster.close()