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
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 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",
])
def test_upload_accepts_ordinary_filenames(name):
    """The allowlist must not break normal use."""
    assert _safe_name_re().match(name), f"should be accepted: {name!r}"


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 = {"shared_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 entirely.
    """
    victim = _session(tmp_path, "victim-user")
    shared_root = victim._ctx["shared_root"]

    original = shared_root / "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"
    uploaded = shared_root / ".uploads" / "attacker-user" / "important.mp4"
    assert uploaded.exists(), "upload should be quarantined, not dropped"
    assert uploaded.read_bytes() == b"attacker content"


def test_upload_rejects_out_of_order_chunks(tmp_path):
    """C5a: chunk_index > 0 used to append blindly to any .part file on disk."""
    session = _session(tmp_path, "user-1")
    session._do_file_upload({
        "filename": "movie.mp4", "chunk_index": 3, "total_chunks": 5,
        "data": base64.b64encode(b"spliced").decode(),
    })
    assert any(m.get("type") == "error" for m in session.sent)


def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path):
    """C5a: even the original uploader goes through a fresh name, not an 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))
    assert any(m.get("type") == "error" for m in session.sent)
    stored = session._ctx["shared_root"] / ".uploads" / "user-1" / "movie.mp4"
    assert stored.read_bytes() == b"first"


# ── 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, "shared_root": tmp_path},
        "b" * 32: {"chat_store": "STORE_B", "index": index_b, "shared_root": tmp_path},
    }
    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()
    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_gek_bundle_store_requires_admin_challenge(tmp_path):
    """
    C5b: gek_bundle_store used to write whatever any authenticated member sent.
    It must now answer with a challenge and store nothing until a valid
    node-operator signature arrives.
    """
    session = _session(tmp_path, "ordinary-member")
    session._group_id = None
    session._admin_ops = {}
    session._ctx["admin_pk_ed25519"] = Ed25519PrivateKey.generate().public_key()

    stored = []

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

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

    assert stored == [], "bundle written without operator authorization (C5b)"
    assert any(m.get("type") == "admin_challenge" for m in session.sent)


def test_gek_bundle_store_refused_without_pinned_admin_key(tmp_path):
    """C5b: deny by default — no pinned key means no privileged operation."""
    session = _session(tmp_path, "ordinary-member")
    session._group_id = None
    session._admin_ops = {}
    session._ctx["bundle_store"] = object()

    session._do_gek_bundle_store({
        "user_id": "victim", "group_id": "g" * 32,
        "pk_eph_b64": "AA==", "nonce_b64": "AA==", "wrapped_b64": "AA==",
    })
    assert any(m.get("type") == "error" for m in session.sent)


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()
    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", "gek_bundle_store"),
    ("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_GEK_BUNDLE_STORE

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

    store_transcript = _transcript(op=OP_GEK_BUNDLE_STORE)
    with pytest.raises(Exception):
        sk_admin.public_key().verify(signature, store_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_admin_ui_escapes_filenames(tmp_path):
    """
    H2: filenames are chosen by any group member and were rendered into the
    localhost admin UI unescaped, giving script execution against an
    unauthenticated admin API.
    """
    from meshbay_node.ui.app import _render_page

    payload = '<img src=x onerror="fetch(1)">'
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    index.add_entry(IndexEntry(
        id="0" * 64, name=payload, path="", size=1, type="video", added_at=0,
    ))

    html = _render_page({
        "status": "running",
        "groups_ctx": {"g" * 32: {"index": index, "shared_root": tmp_path}},
        "indexes": {"g" * 32: index},
    })

    assert payload not in html, "filename rendered unescaped — stored XSS (H2)"
    assert "&lt;img" in html, "filename should appear escaped"