From da89eadb3dfddef7eaf8166a8353a20e62bbf809 Mon Sep 17 00:00:00 2001 From: Ivan Date: Thu, 13 Aug 2026 14:33:24 -0500 Subject: [PATCH] refactor: update landlock sandbox path normalization and improve repository server error handling in demo mode --- meshchatx.rsm | Bin 191718 -> 191718 bytes meshchatx/meshchat.py | 3 ++ .../backend/http/routes/repository_server.py | 3 ++ .../backend/http/routes/websocket_upgrade.py | 13 ++++++ meshchatx/src/backend/landlock_sandbox.py | 37 ++++++++++++++++-- tests/backend/test_app_security_features.py | 18 +++++++++ .../test_landlock_integration_surfaces.py | 13 +++++- tests/backend/test_landlock_sandbox.py | 23 +++++++++++ tests/backend/test_rns_lifecycle.py | 30 ++++++++++++++ tests/backend/test_telephone_audio_ws.py | 22 +++++++++++ 10 files changed, 157 insertions(+), 5 deletions(-) diff --git a/meshchatx.rsm b/meshchatx.rsm index 6190be2edcee765c4d99a17185a46c8ee9452f7a..ff854582ca1451f328435c8fbd9ce398d000abfe 100644 GIT binary patch delta 792 zcmX|9O^6j#6vkKN&rHH>CVv(+3qc*qd+s^so^v7QJ!lAQ6^Jqlk@x4^b|r|Q?!=!} zpyMOnqQ$g`7Ojd=9<PjIOs+qA;Cp$S~O%=f;j8%|2*5AdA2z-g8SF_=YzLC zJN8=IT70ZCv)9h(l}D%I@?U>nxc>3R$u~~#{q*8bFLoXJ{oH#8{@hqUdTIHWeM`q+ z?whA8r`snxS4Vr^dH>rFx2}Hq^628dmA4oFSoz`d_urj7uzB-~=l4%Mc_3Ryems9S zje-#d7m}3B5Pa2&1i;NvP_$OLBqqm_xtd69)mW%fXff3oLtej-AH6$F9%^niHWX)2 zP)#$FmXbn~;xc4~D2`l=*g*|r$<-Hr5dm}}31U+bE4$4;h(|Ak{{j1f5QV6XzE2Y@v P&W^!b)BVQ>8n=&e4{HvOh-_;7yw>CoL>&kmeXEgAI}aByeTPnxu{zO(;EXQ(0nsL{ z!8jBUoJE0<$q9@ZB~9J9vv1+=^!rVXBpI!xmXYTNCp}qWoJG>o7g?fE=uLqvC@~OY zB0eY~S!V^})Z}RQME`87f3|JucwJ`gGHL*BAOnL%GJ#MDr$aQ92}{bX&?o}3kR*7X z!Y>x04%ED~)>dr|d~W-*zh;Qs`a4@>PzACsq-2xAoUGBT*$0)CiqS`5qgFnIB9fHF zdy#E(TVJKCIkTm*r=GbcyI`^jf_G)CMC~QZKvjr$X`RwSiRg;+!BWgV72ZgMEGYvt zhXyOv^-dkF9a}4lhw2`K^<*Mhmggc6TP^`|f{dsvM}(PSG0A1Ck<%bUa)CMV1sVTq z6t&8OYAsl)h}R@}yZ|Psa}tTu!!Ac=>`7pbI*Z`hM?_CvbH str | None: + """Return a realpath extra read root, or None when it would widen the jail. + + Rejects the filesystem root and the user home directory itself. A Sideband + plugins folder may live under home. Home or / as the extra root would let + a compromised plugin read ssh keys and the rest of the host tree. + """ + if not isinstance(path, str) or not path.strip(): + return None + try: + resolved = os.path.realpath(os.path.abspath(os.path.expanduser(path.strip()))) + except OSError: + return None + if not os.path.isdir(resolved): + return None + fs_root = os.path.realpath(os.path.abspath(os.sep)) + if resolved == fs_root: + return None + home = os.path.expanduser("~") + if home and home != "~": + try: + home_real = os.path.realpath(os.path.abspath(home)) + except OSError: + home_real = "" + if home_real and resolved == home_real: + return None + return resolved + + def extra_read_roots_from_app(app) -> list[str]: """Sideband command-plugin dirs chosen in settings, if they exist on disk. @@ -505,8 +534,8 @@ def extra_read_roots_from_app(app) -> list[str]: raw = None if not raw: return [] - resolved = os.path.abspath(os.path.expanduser(str(raw))) - if os.path.isdir(resolved): + resolved = _normalize_extra_landlock_read_root(str(raw)) + if resolved: return [resolved] return [] @@ -562,8 +591,8 @@ def apply_landlock_sandbox( for extra in extra_read_roots or []: if not extra: continue - resolved = os.path.abspath(os.path.expanduser(str(extra))) - if os.path.isdir(resolved) and resolved not in read_roots: + resolved = _normalize_extra_landlock_read_root(str(extra)) + if resolved and resolved not in read_roots: read_roots.append(resolved) for root in read_roots: _add_path_beneath_rule( diff --git a/tests/backend/test_app_security_features.py b/tests/backend/test_app_security_features.py index c63b3995..16f555ac 100644 --- a/tests/backend/test_app_security_features.py +++ b/tests/backend/test_app_security_features.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: 0BSD import secrets +from unittest.mock import MagicMock import bcrypt import pytest @@ -153,3 +154,20 @@ async def test_privacy_mode_blocks_map_export(mock_app): headers=headers, ) assert r.status == 403 + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("require_loopback_tcp") +async def test_privacy_mode_blocks_repository_refresh(mock_app): + mock_app.config.privacy_mode_enabled.set(True) + mock_app.repository_server_manager = MagicMock() + aio_app = _make_aio_app(mock_app, use_https=False) + + async with TestClient(TestServer(aio_app)) as client: + headers = await fetch_api_csrf_headers(client) + r = await client.post( + "/api/v1/repository-server/refresh-bundled", + headers=headers, + ) + assert r.status == 403 + mock_app.repository_server_manager.refresh_bundled_wheels.assert_not_called() diff --git a/tests/backend/test_landlock_integration_surfaces.py b/tests/backend/test_landlock_integration_surfaces.py index 0d1ddbf7..51a06ae5 100644 --- a/tests/backend/test_landlock_integration_surfaces.py +++ b/tests/backend/test_landlock_integration_surfaces.py @@ -603,5 +603,16 @@ def test_extra_read_roots_from_app_uses_command_plugins_path(tmp_path): assert ll.extra_read_roots_from_app(_App(str(missing))) == [] present = tmp_path / "plugins" present.mkdir() - assert ll.extra_read_roots_from_app(_App(str(present))) == [str(present)] + assert ll.extra_read_roots_from_app(_App(str(present))) == [ + os.path.realpath(str(present)), + ] assert ll.extra_read_roots_from_app(_App(None)) == [] + assert ll.extra_read_roots_from_app(_App(os.sep)) == [] + home = os.path.expanduser("~") + if home and home != "~" and os.path.isdir(home): + assert ll.extra_read_roots_from_app(_App(home)) == [] + nested = tmp_path / "homeish" / "plugins" + nested.mkdir(parents=True) + assert ll.extra_read_roots_from_app(_App(str(nested))) == [ + os.path.realpath(str(nested)), + ] diff --git a/tests/backend/test_landlock_sandbox.py b/tests/backend/test_landlock_sandbox.py index 52ae2e29..c4284799 100644 --- a/tests/backend/test_landlock_sandbox.py +++ b/tests/backend/test_landlock_sandbox.py @@ -313,3 +313,26 @@ def test_apply_landlock_allows_user_local_argospm_list(tmp_path): storage=storage, ) assert_probe_ok(result) + + +def test_normalize_extra_landlock_read_root_rejects_overbroad(tmp_path): + plugins = tmp_path / "sideband-plugins" + plugins.mkdir() + assert ll._normalize_extra_landlock_read_root(str(plugins)) == os.path.realpath( + str(plugins), + ) + assert ll._normalize_extra_landlock_read_root(os.sep) is None + assert ll._normalize_extra_landlock_read_root("") is None + home = os.path.expanduser("~") + if home and home != "~" and os.path.isdir(home): + assert ll._normalize_extra_landlock_read_root(home) is None + + +@pytest.mark.skipif(os.name == "nt", reason="symlink semantics differ on Windows") +def test_normalize_extra_landlock_read_root_rejects_symlink_to_fs_root(tmp_path): + link = tmp_path / "rootlink" + try: + link.symlink_to(os.sep) + except OSError: + pytest.skip("cannot create symlink to filesystem root") + assert ll._normalize_extra_landlock_read_root(str(link)) is None diff --git a/tests/backend/test_rns_lifecycle.py b/tests/backend/test_rns_lifecycle.py index 52bebba5..48d29807 100644 --- a/tests/backend/test_rns_lifecycle.py +++ b/tests/backend/test_rns_lifecycle.py @@ -910,3 +910,33 @@ async def test_reload_teardown_stops_all_context_services(mock_rns, temp_dir): ctx_b.teardown.assert_called_once() assert app.contexts == {} assert app.current_context is None + + +def test_teardown_all_contexts_for_reload_clears_mesh_link_caches(mock_rns, temp_dir): + with ( + patch("meshchatx.src.backend.identity_context.Database"), + patch("meshchatx.src.backend.identity_context.ConfigManager"), + patch("meshchatx.src.backend.identity_context.MessageHandler"), + patch("meshchatx.src.backend.identity_context.AnnounceManager"), + patch("meshchatx.src.backend.identity_context.ArchiverManager"), + patch("meshchatx.src.backend.identity_context.MapManager"), + patch("meshchatx.src.backend.identity_context.TelephoneManager"), + patch("meshchatx.src.backend.identity_context.VoicemailManager"), + patch("meshchatx.src.backend.identity_context.RingtoneManager"), + patch("meshchatx.src.backend.identity_context.RNCPHandler"), + patch("meshchatx.src.backend.identity_context.RNStatusHandler"), + patch("meshchatx.src.backend.identity_context.RNProbeHandler"), + patch("meshchatx.src.backend.identity_context.TranslatorHandler"), + patch("LXMF.LXMRouter"), + ): + app = ReticulumMeshChat( + identity=mock_rns["id_instance"], + storage_dir=temp_dir, + reticulum_config_dir=temp_dir, + ) + app.contexts = {} + app.current_context = None + app.page_node_manager.teardown = MagicMock() + app._clear_mesh_link_caches = MagicMock() + app._teardown_all_contexts_for_reload() + app._clear_mesh_link_caches.assert_called_once() diff --git a/tests/backend/test_telephone_audio_ws.py b/tests/backend/test_telephone_audio_ws.py index 83f4a28d..5b2a01cd 100644 --- a/tests/backend/test_telephone_audio_ws.py +++ b/tests/backend/test_telephone_audio_ws.py @@ -192,3 +192,25 @@ async def test_telephone_audio_ws_bad_json_does_not_crash_handler(web_audio_app) assert pong["type"] == "pong" bridge.detach_client.assert_called_once() + + +@pytest.mark.asyncio +async def test_telephone_audio_ws_demo_mode_rejects_before_attach(web_audio_app): + from meshchatx.src.backend.demo_mode import DEMO_READONLY_CODE + + web_audio_app.demo_mode = True + bridge = _bridge_with_clients() + web_audio_app.web_audio_bridge = bridge + + aio_app = _build_aio_app(web_audio_app) + async with TestClient(TestServer(aio_app)) as client: + ws = await client.ws_connect("/ws/telephone/audio") + msg = await ws.receive_json() + await ws.send_bytes(b"\x01\x02\x03") + await ws.close() + + assert msg["type"] == "error" + assert msg["code"] == DEMO_READONLY_CODE + bridge.send_status.assert_not_called() + bridge.attach_client.assert_not_called() + bridge.push_client_frame.assert_not_called()