1 files changed, 372 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
new file mode 100644
index 0000000..a480d21
--- /dev/null
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -0,0 +1,372 @@
+"""
+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"