From 328b01a2dd545d70a078db8df1e91b02d65bfc9c Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Wed, 23 Sep 2026 22:25:28 +0200 Subject: test: read the WebRTC transport's source as a set of files Source-reading tests take their text from node_source (node) and node_tree (hub): webrtc_server.py plus anything under transport/webrtc/, so a check for something's absence keeps reading the code it guards if that code moves. test_node_source_scope holds the boundary. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_security_regressions.py | 25 ++++++++++------------ 1 file changed, 11 insertions(+), 14 deletions(-) (limited to 'packages/meshbay-node/tests/test_security_regressions.py') diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 7010523..7ace21b 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -18,6 +18,7 @@ from meshbay_common.crypto import generate_gek from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import RootSet from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from node_source import TRANSPORT, webrtc_files, webrtc_source from conftest import one_root, opened_ack, sealed_upload @@ -461,9 +462,7 @@ def test_no_member_can_hand_the_node_key_material(tmp_path): "key material over MNP (C5b)" ) - source = (Path(__file__).parent.parent - / "src" / "meshbay_node" / "transport" - / "webrtc_server.py").read_text(encoding="utf-8") + source = webrtc_source() assert "_do_gek_bundle_store" not in source assert "_admin_exec_bundle_store" not in source @@ -496,8 +495,7 @@ def test_gek_auto_activation_is_gone(): 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") + source = webrtc_source() 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" @@ -709,8 +707,7 @@ def test_peer_errors_do_not_leak_internals(): 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") + source = webrtc_source() assert '"detail": str(e)' not in source, ( "generic exception text relayed to peer — use a fixed message" ) @@ -759,9 +756,10 @@ def test_no_transport_ships_media_outside_the_aead(): "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") + quic = [TRANSPORT / "quic_server.py", TRANSPORT / "quic_client.py"] + for path in webrtc_files() + quic: + name = path.name + source = path.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), ( @@ -780,11 +778,10 @@ def test_ffmpeg_never_blocks_the_event_loop(): """ import ast - source = (Path(__file__).parent.parent / "src" / "meshbay_node" - / "transport" / "webrtc_server.py").read_text(encoding="utf-8") - tree = ast.parse(source) + source = webrtc_source() blocking = [ - node for node in ast.walk(tree) + node for path in webrtc_files() + for node in ast.walk(ast.parse(path.read_text(encoding="utf-8"))) if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "run" -- cgit v1.2.3