1 files changed, 75 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 9552848..a6d69c6 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -10,6 +10,7 @@ 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
@@ -465,6 +466,80 @@ def test_peer_errors_do_not_leak_internals():
     assert '"detail": str(e)' not 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()
+    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/config?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_admin_ui_escapes_filenames(tmp_path):
     """
     H2: filenames are chosen by any group member and were rendered into the