From 15cc6926a4d2affd617222cb2c353a460c954b3b Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Thu, 17 Sep 2026 08:48:21 -0400 Subject: [PATCH] fix(web): keep plugin asset and action routes inside their directories POST /plugins/assets/upload, GET /plugins/assets/list and POST /plugins/assets/delete joined the request's plugin_id onto assets/plugins unchecked, so '../../config' created, wrote, listed and deleted outside it. #561 guarded only the route that serves the files. All three now go through path_safety.resolve_under and answer 400 for anything but a plain name, and delete only unlinks a metadata path that resolves into that plugin's uploads directory. PluginManager.get_plugin_directory refuses ids that are not one plain path segment, so /plugins/action (which runs a manifest script from the returned directory) and every other caller get the guard; the action route also rejects such ids up front, covering its no-manager fallback. Co-Authored-By: Claude Opus 5 --- src/plugin_system/plugin_manager.py | 12 +- test/test_path_traversal_guards.py | 199 +++++++++++++++++++++ web_interface/blueprints/api_v3/plugins.py | 46 ++++- 3 files changed, 249 insertions(+), 8 deletions(-) diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py index c587acc8..a79ec867 100644 --- a/src/plugin_system/plugin_manager.py +++ b/src/plugin_system/plugin_manager.py @@ -23,6 +23,7 @@ from src.plugin_system.plugin_loader import PluginLoader from src.plugin_system.plugin_executor import PluginExecutor from src.plugin_system.plugin_state import PluginStateManager, PluginState from src.plugin_system.schema_manager import SchemaManager, normalize_legacy_booleans +from src.common.path_safety import safe_path_component from src.common.permission_utils import ( ensure_directory_permissions, get_plugin_dir_mode @@ -743,11 +744,20 @@ class PluginManager: Returns: Directory path as string or None if not found + + ``plugin_id`` often comes straight from a request, so anything that is + not one plain path segment (``..``, ``a/b``, an absolute path) is + refused instead of being joined onto ``plugins_dir``. The join is not + resolved further: dev plugins are symlinks into ``plugins_dir``. """ with self._discovery_lock: if hasattr(self, 'plugin_directories') and plugin_id in self.plugin_directories: return str(self.plugin_directories[plugin_id]) - + + plugin_id = safe_path_component(plugin_id) + if plugin_id is None: + return None + plugin_dir = self.plugins_dir / plugin_id if plugin_dir.exists(): return str(plugin_dir) diff --git a/test/test_path_traversal_guards.py b/test/test_path_traversal_guards.py index 9aaa04e4..bce107cc 100644 --- a/test/test_path_traversal_guards.py +++ b/test/test_path_traversal_guards.py @@ -315,3 +315,202 @@ class TestPluginVersionLookup: assert module._get_plugin_version("..") == "" finally: module.api_v3.plugin_store_manager = original + + +PNG = b"\x89PNG\r\n\x1a\n" + b"0" * 20 + + +class TestPluginAssetRoutes: + """/api/v3/plugins/assets/upload, /list and /delete + + All three joined a request-supplied plugin_id straight onto + assets/plugins, so '../../config' created an uploads directory under + config/, wrote images and .metadata.json there, listed it, and deleted + whatever file a metadata entry's path named. #561 fixed only the route + that serves the uploaded files. + """ + + @pytest.fixture + def project(self, tmp_path, api_v3_module, monkeypatch): + import web_interface.blueprints.api_v3.plugins as plugins_module + + monkeypatch.setattr(plugins_module, "PROJECT_ROOT", tmp_path) + (tmp_path / "config").mkdir() + secrets = tmp_path / "config" / "config_secrets.json" + secrets.write_text('{"api_key":"hunter2"}', encoding="utf-8") + return tmp_path, secrets + + def _upload(self, client, plugin_id): + import io + + return client.post( + "/api/v3/plugins/assets/upload", + data={"plugin_id": plugin_id, "files": (io.BytesIO(PNG), "x.png")}, + content_type="multipart/form-data", + ) + + def _tree(self, root): + return sorted(str(p.relative_to(root)) for p in root.rglob("*")) + + def test_an_ordinary_upload_list_and_delete_still_work(self, api_v3_client, project): + root, _ = project + response = self._upload(api_v3_client, "static-image") + assert response.status_code == 200, response.get_json() + uploaded = response.get_json()["uploaded_files"][0] + assert uploaded["path"].startswith("assets/plugins/static-image/uploads/") + stored = root / uploaded["path"] + assert stored.exists() + + listed = api_v3_client.get( + "/api/v3/plugins/assets/list", query_string={"plugin_id": "static-image"} + ) + assert listed.status_code == 200 + assert [a["id"] for a in listed.get_json()["data"]["assets"]] == [uploaded["id"]] + + deleted = api_v3_client.post( + "/api/v3/plugins/assets/delete", + json={"plugin_id": "static-image", "image_id": uploaded["id"]}, + ) + assert deleted.status_code == 200 + assert not stored.exists() + + @pytest.mark.parametrize("plugin_id", [ + "../../config", "..", "a/b", "/abs", "x" + BACKSLASH + "y", "C:", + ]) + def test_a_traversing_upload_writes_nothing(self, api_v3_client, project, plugin_id): + root, _ = project + before = self._tree(root) + response = self._upload(api_v3_client, plugin_id) + assert response.status_code == 400 + assert self._tree(root) == before + + def test_a_traversing_list_reads_nothing(self, api_v3_client, project): + root, _ = project + outside = root / "config" / "uploads" + outside.mkdir() + (outside / ".metadata.json").write_text( + json.dumps({"i": {"id": "i", "path": "leaked"}}), encoding="utf-8" + ) + response = api_v3_client.get( + "/api/v3/plugins/assets/list", query_string={"plugin_id": "../../config"} + ) + assert response.status_code == 400 + assert b"leaked" not in response.data + + def test_a_traversing_delete_deletes_nothing(self, api_v3_client, project): + root, secrets = project + outside = root / "config" / "uploads" + outside.mkdir() + (outside / ".metadata.json").write_text( + json.dumps({"i": {"id": "i", "path": "config/config_secrets.json"}}), + encoding="utf-8", + ) + response = api_v3_client.post( + "/api/v3/plugins/assets/delete", + json={"plugin_id": "../../config", "image_id": "i"}, + ) + assert response.status_code == 400 + assert secrets.exists(), "file outside assets/plugins was deleted" + + @pytest.mark.parametrize("stored_path", [ + "config/config_secrets.json", + "assets/plugins/static-image/uploads/../../../../config/config_secrets.json", + "assets/plugins/other/uploads/x.png", + None, + ]) + def test_a_metadata_path_outside_the_uploads_is_never_unlinked( + self, api_v3_client, project, stored_path + ): + root, secrets = project + uploads = root / "assets" / "plugins" / "static-image" / "uploads" + uploads.mkdir(parents=True) + other = root / "assets" / "plugins" / "other" / "uploads" + other.mkdir(parents=True) + (other / "x.png").write_bytes(PNG) + (uploads / ".metadata.json").write_text( + json.dumps({"i": {"id": "i", "path": stored_path}}), encoding="utf-8" + ) + response = api_v3_client.post( + "/api/v3/plugins/assets/delete", + json={"plugin_id": "static-image", "image_id": "i"}, + ) + assert response.status_code == 200 + assert secrets.exists(), "file named by tampered metadata was deleted" + assert (other / "x.png").exists(), "another plugin's upload was deleted" + # The bad entry is still dropped, so the UI can get rid of it. + assert json.loads((uploads / ".metadata.json").read_text(encoding="utf-8")) == {} + + +class TestPluginActionDirectory: + """POST /api/v3/plugins/action runs a script named by the manifest in + get_plugin_directory(plugin_id), and plugin_id is the request body's. + get_plugin_directory joined it onto plugins_dir unchecked, so + '../elsewhere' ran a script from any directory holding a manifest. + """ + + @pytest.fixture + def tree(self, tmp_path): + import threading + + from src.plugin_system.plugin_manager import PluginManager + + plugins = tmp_path / "plugin-repos" + (plugins / "demo").mkdir(parents=True) + (plugins / "ledmatrix-legacy").mkdir() + outside = tmp_path / "elsewhere" + outside.mkdir() + (outside / "manifest.json").write_text(json.dumps({ + "web_ui_actions": [{"id": "go", "type": "script", "script": "s.py"}], + }), encoding="utf-8") + marker = tmp_path / "PWNED" + (outside / "s.py").write_text( + "open(%r, 'w').write('ran')\n" % str(marker), encoding="utf-8" + ) + manager = MagicMock() + manager.plugins_dir = plugins + manager._discovery_lock = threading.Lock() + manager.plugin_directories = {} + manager.get_plugin_directory = ( + lambda pid: PluginManager.get_plugin_directory(manager, pid) + ) + return manager, plugins, marker + + def test_real_plugin_directories_are_still_found(self, tree): + manager, plugins, _ = tree + assert manager.get_plugin_directory("demo") == str(plugins / "demo") + assert manager.get_plugin_directory("legacy") == str(plugins / "ledmatrix-legacy") + assert manager.get_plugin_directory("nope") is None + + def test_a_discovered_plugin_is_returned_from_the_registry(self, tree, tmp_path): + manager, _, _ = tree + manager.plugin_directories = {"linked": tmp_path / "somewhere"} + assert manager.get_plugin_directory("linked") == str(tmp_path / "somewhere") + + @pytest.mark.parametrize("plugin_id", [ + "..", "../elsewhere", "a/b", "/abs", "x" + BACKSLASH + "..", "", + ]) + def test_get_plugin_directory_refuses_anything_but_a_plain_name(self, tree, plugin_id): + manager, _, _ = tree + assert manager.get_plugin_directory(plugin_id) is None + + def test_the_action_endpoint_runs_nothing_outside_the_plugins_dir( + self, api_v3_client, api_v3_module, tree + ): + manager, _, marker = tree + api_v3_module.api_v3.plugin_manager = manager + response = api_v3_client.post( + "/api/v3/plugins/action", + json={"plugin_id": "../elsewhere", "action_id": "go", "params": {}}, + ) + assert response.status_code == 400 + assert not marker.exists(), "script outside the plugins directory ran" + + def test_the_fallback_without_a_plugin_manager_is_guarded_too( + self, api_v3_client, api_v3_module + ): + api_v3_module.api_v3.plugin_manager = None + response = api_v3_client.post( + "/api/v3/plugins/action", + json={"plugin_id": "../elsewhere", "action_id": "go", "params": {}}, + ) + assert response.status_code == 400 diff --git a/web_interface/blueprints/api_v3/plugins.py b/web_interface/blueprints/api_v3/plugins.py index c9643e54..87d1cbd2 100644 --- a/web_interface/blueprints/api_v3/plugins.py +++ b/web_interface/blueprints/api_v3/plugins.py @@ -2832,6 +2832,12 @@ def execute_plugin_action(): 'received': {'plugin_id': plugin_id, 'action_id': action_id, 'has_params': bool(action_params)} }), 400 + # plugin_id comes from the request body and picks the directory whose + # manifest names the script to run, so it must be a plain name. + plugin_id = safe_path_component(plugin_id) + if plugin_id is None: + return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400 + # Get plugin directory if api_v3.plugin_manager: plugin_dir = api_v3.plugin_manager.get_plugin_directory(plugin_id) @@ -3303,6 +3309,15 @@ def authenticate_ytm(): except Exception as e: logger.error('Error in authenticate_ytm', exc_info=True) return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500 +def _plugin_uploads_dir(plugin_id): + """assets/plugins//uploads for a request-supplied id, or None. + + Same guard as the serving route in app.py: one plain path segment, + resolved and contained under assets/plugins. + """ + return resolve_under(PROJECT_ROOT / 'assets' / 'plugins', plugin_id, 'uploads') + + @api_v3.route('/plugins/assets/upload', methods=['POST']) def upload_plugin_asset(): """Upload asset files for a plugin""" @@ -3322,8 +3337,13 @@ def upload_plugin_asset(): if len(files) > 10: return jsonify({'status': 'error', 'message': 'Maximum 10 files per upload'}), 400 - # Setup plugin assets directory - assets_dir = PROJECT_ROOT / 'assets' / 'plugins' / plugin_id / 'uploads' + # Setup plugin assets directory. plugin_id is a form field: without + # the guard '../../config' created, listed and wrote into directories + # outside assets/plugins (the serving route was fixed in #561). + assets_dir = _plugin_uploads_dir(plugin_id) + if assets_dir is None: + return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400 + plugin_id = safe_path_component(plugin_id) assets_dir.mkdir(parents=True, exist_ok=True) # Load metadata file @@ -3958,7 +3978,9 @@ def delete_plugin_asset(): return jsonify({'status': 'error', 'message': 'plugin_id and image_id are required'}), 400 # Get asset directory - assets_dir = PROJECT_ROOT / 'assets' / 'plugins' / plugin_id / 'uploads' + assets_dir = _plugin_uploads_dir(plugin_id) + if assets_dir is None: + return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400 metadata_file = assets_dir / '.metadata.json' if not metadata_file.exists(): @@ -3971,9 +3993,17 @@ def delete_plugin_asset(): if image_id not in metadata: return jsonify({'status': 'error', 'message': 'Image not found'}), 404 - # Delete file - file_path = PROJECT_ROOT / metadata[image_id]['path'] - if file_path.exists(): + # Delete file. The stored path is data, not a trusted location: only + # unlink it when it resolves to a file directly inside this plugin's + # uploads. An entry pointing anywhere else is dropped from the + # metadata without touching the file it names. + entry = metadata[image_id] if isinstance(metadata[image_id], dict) else {} + parts = safe_relative_parts(entry.get('path')) + file_path = resolve_under(PROJECT_ROOT, *parts) if parts else None + if file_path is None or file_path.parent != assets_dir: + logger.warning('Asset %s has a path outside its uploads directory; ' + 'removing the entry without deleting a file', image_id) + elif file_path.exists(): file_path.unlink() # Remove from metadata @@ -3997,7 +4027,9 @@ def list_plugin_assets(): return jsonify({'status': 'error', 'message': 'plugin_id is required'}), 400 # Get asset directory - assets_dir = PROJECT_ROOT / 'assets' / 'plugins' / plugin_id / 'uploads' + assets_dir = _plugin_uploads_dir(plugin_id) + if assets_dir is None: + return jsonify({'status': 'error', 'message': 'Invalid plugin_id'}), 400 metadata_file = assets_dir / '.metadata.json' if not metadata_file.exists():