From bbe2a6312703e7bfe5c44aaeae2f02de4f1649fa Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Aug 2026 14:01:52 +0000 Subject: [PATCH] test(api): cover backup restore and path containment, and fix restore scope MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restore is the most destructive thing the web interface can do — it overwrites config, secrets, WiFi settings and fonts, then reinstalls plugins — and neither it nor the file routes beside it had tests. A malformed `options` field fell back to {}. Every RestoreOptions flag defaults to True, so a caller who asked for a narrow restore and mis-serialized the request got a full one instead, secrets included, and was told it succeeded. Valid JSON that is not an object was worse: `"null"` or `"[1,2]"` reached .get() on a non-dict and raised, so the request died as a generic 500. Both are now refused with a 400 that says what was wrong, and restore_backup is never reached. The other file routes take a filename straight out of the URL and turn it into a path — one to read, one to unlink. _safe_backup_path is the only thing keeping those inside the export directory, and it was untested. No bypass was found; the thirteen traversal shapes are pinned so a later loosening of that pattern has to argue with something. The delete route's by-name enumeration is covered too, including that a directory sharing a backup's name is not removed. 84 tests. Two behaviours are pinned as intentional: a failed plugin reinstall turns the whole restore into an error even though file restoration succeeded, and omitting `options` entirely still means restore everything — that is the documented default, and it is only the mis-serialized case that was wrong. Co-Authored-By: Claude Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh --- .../web_interface/test_api_v3_backup_paths.py | 220 +++++++++++++++ .../test_api_v3_backup_restore.py | 262 ++++++++++++++++++ web_interface/blueprints/api_v3.py | 11 +- 3 files changed, 492 insertions(+), 1 deletion(-) create mode 100644 test/web_interface/test_api_v3_backup_paths.py create mode 100644 test/web_interface/test_api_v3_backup_restore.py diff --git a/test/web_interface/test_api_v3_backup_paths.py b/test/web_interface/test_api_v3_backup_paths.py new file mode 100644 index 00000000..73dc47d1 --- /dev/null +++ b/test/web_interface/test_api_v3_backup_paths.py @@ -0,0 +1,220 @@ +""" +Path-containment tests for the backup file routes: +GET /backup/download/, DELETE /backup/, and the +listing/validation routes alongside them. + +Both filename routes take user input straight from the URL and turn it +into a filesystem path, one to read and one to unlink. `_safe_backup_path` +is what stops that from reaching outside the export directory, and it had +no tests. + +This is verification of existing containment, not a fix: no bypass was +found. The tests exist so that a later "just let dots through" change has +to argue with something. +""" + +import io +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from flask import Flask + +project_root = Path(__file__).parent.parent.parent +sys.path.insert(0, str(project_root)) + +from web_interface.blueprints import api_v3 as api_v3_module # noqa: E402 +from web_interface.blueprints.api_v3 import api_v3 # noqa: E402 + +_MANAGER_ATTRS = ( + 'config_manager', 'plugin_manager', 'plugin_store_manager', + 'plugin_state_manager', 'saved_repositories_manager', 'schema_manager', + 'operation_queue', 'operation_history', 'cache_manager', +) +_SENTINEL = object() + +# Anything that tries to name a file outside the export directory, or that +# is not a plain .zip. +TRAVERSAL_ATTEMPTS = [ + "../../etc/passwd", + "../config.json", + "..%2f..%2fetc%2fpasswd", + "....//....//etc/passwd", + "/etc/passwd", + "..\\..\\config.json", + "backup.zip/../../../etc/passwd", + ".hidden.zip", + "backup.txt", + "backup.zip.exe", + "", + ".", + "..", +] + + +@pytest.fixture +def env(tmp_path, monkeypatch): + export_dir = tmp_path / "backups" + export_dir.mkdir() + monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", export_dir) + + # A file outside the export dir that a traversal would be reaching for. + secret = tmp_path / "config.json" + secret.write_text(json.dumps({"secret": "do not touch"})) + + originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS} + for name in _MANAGER_ATTRS: + setattr(api_v3, name, MagicMock()) + + app = Flask(__name__) + app.config["TESTING"] = True + app.register_blueprint(api_v3, url_prefix="/api/v3") + + class Env: + pass + + e = Env() + e.client = app.test_client() + e.export_dir = export_dir + e.secret = secret + yield e + + for name, original in originals.items(): + if original is _SENTINEL: + if hasattr(api_v3, name): + delattr(api_v3, name) + else: + setattr(api_v3, name, original) + + +def make_backup(export_dir, name="backup-2026-01-01.zip"): + path = export_dir / name + path.write_bytes(b"PK\x03\x04fake zip") + return path + + +class TestSafeBackupPath: + """The containment helper itself.""" + + @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS) + def test_rejects_unsafe_names(self, env, filename): + assert api_v3_module._safe_backup_path(filename) is None + + def test_rejects_none(self, env): + assert api_v3_module._safe_backup_path(None) is None + + @pytest.mark.parametrize("filename", [ + "backup.zip", + "backup-2026-01-01.zip", + "backup_2026.01.01-v2.zip", + "a.zip", + ]) + def test_accepts_plain_zip_names(self, env, filename): + resolved = api_v3_module._safe_backup_path(filename) + assert resolved is not None + assert resolved.parent == env.export_dir.resolve() + + def test_result_is_always_inside_the_export_dir(self, env): + resolved = api_v3_module._safe_backup_path("backup.zip") + resolved.relative_to(env.export_dir.resolve()) # raises if outside + + def test_overlong_name_rejected(self, env): + assert api_v3_module._safe_backup_path("a" * 250 + ".zip") is None + + +class TestDownload: + def test_downloads_an_existing_backup(self, env): + make_backup(env.export_dir) + response = env.client.get("/api/v3/backup/download/backup-2026-01-01.zip") + assert response.status_code == 200 + assert response.data == b"PK\x03\x04fake zip" + + def test_missing_file_is_a_404(self, env): + response = env.client.get("/api/v3/backup/download/never-made.zip") + assert response.status_code == 404 + + @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS) + def test_traversal_attempts_are_refused(self, env, filename): + response = env.client.get(f"/api/v3/backup/download/{filename}") + # However the request is turned away — 404 from the containment + # check, or 308/405 from routing never matching at all — what + # matters is that no file outside the export directory is served. + assert response.status_code != 200 + assert b"do not touch" not in response.data + + +class TestDelete: + def test_deletes_an_existing_backup(self, env): + path = make_backup(env.export_dir) + response = env.client.delete("/api/v3/backup/backup-2026-01-01.zip") + assert response.status_code == 200 + assert not path.exists() + + def test_missing_file_is_a_404(self, env): + response = env.client.delete("/api/v3/backup/never-made.zip") + assert response.status_code == 404 + + @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS) + def test_traversal_attempts_delete_nothing(self, env, filename): + response = env.client.delete(f"/api/v3/backup/{filename}") + assert response.status_code != 200 + assert env.secret.exists() # the file a traversal was aiming at + + def test_only_the_named_backup_is_removed(self, env): + keep = make_backup(env.export_dir, "keep.zip") + drop = make_backup(env.export_dir, "drop.zip") + env.client.delete("/api/v3/backup/drop.zip") + assert keep.exists() + assert not drop.exists() + + def test_directory_with_a_matching_name_is_not_removed(self, env): + # The delete loop matches by name but requires a regular file. + (env.export_dir / "sneaky.zip").mkdir() + response = env.client.delete("/api/v3/backup/sneaky.zip") + assert response.status_code == 404 + assert (env.export_dir / "sneaky.zip").is_dir() + + +class TestList: + def test_lists_only_zip_files(self, env): + make_backup(env.export_dir, "one.zip") + (env.export_dir / "notes.txt").write_text("ignore me") + response = env.client.get("/api/v3/backup/list") + assert response.status_code == 200 + names = [entry["filename"] for entry in response.get_json()["data"]] + assert names == ["one.zip"] + + def test_empty_directory_lists_nothing(self, env): + response = env.client.get("/api/v3/backup/list") + assert response.get_json()["data"] == [] + + def test_entries_carry_size_and_timestamp(self, env): + make_backup(env.export_dir, "one.zip") + entry = env.client.get("/api/v3/backup/list").get_json()["data"][0] + assert entry["size"] == len(b"PK\x03\x04fake zip") + assert entry["created_at"] + + +class TestValidate: + def test_missing_file_is_a_400(self, env): + response = env.client.post("/api/v3/backup/validate", data={}, + content_type="multipart/form-data") + assert response.status_code == 400 + assert "No backup_file" in response.get_json()["message"] + + def test_invalid_archive_is_a_400(self, env): + response = env.client.post( + "/api/v3/backup/validate", + data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")}, + content_type="multipart/form-data") + assert response.status_code == 400 + assert "Invalid or corrupted" in response.get_json()["message"] + + def test_validation_does_not_leave_temp_files_in_the_export_dir(self, env): + env.client.post( + "/api/v3/backup/validate", + data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")}, + content_type="multipart/form-data") + assert list(env.export_dir.iterdir()) == [] diff --git a/test/web_interface/test_api_v3_backup_restore.py b/test/web_interface/test_api_v3_backup_restore.py new file mode 100644 index 00000000..7b67ca16 --- /dev/null +++ b/test/web_interface/test_api_v3_backup_restore.py @@ -0,0 +1,262 @@ +""" +Endpoint tests for POST /backup/restore. + +Restore is the most destructive operation the web interface exposes: it +overwrites config, secrets, WiFi settings and fonts, and reinstalls +plugins. It had no tests. + +restore_backup itself is mocked — this file is about what the route does +with the request and with the result, not about ZIP handling, which +belongs to backup_manager's own tests. + +Regression coverage for one fixed bug: a malformed `options` field fell +back to {}, and since every RestoreOptions flag defaults to True, that +turned a mis-serialized narrow restore into a full one — secrets +included — with no indication anything had been ignored. +""" + +import io +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from flask import Flask + +project_root = Path(__file__).parent.parent.parent +sys.path.insert(0, str(project_root)) + +from web_interface.blueprints.api_v3 import api_v3 # noqa: E402 + +URL = "/api/v3/backup/restore" + +_MANAGER_ATTRS = ( + 'config_manager', 'plugin_manager', 'plugin_store_manager', + 'plugin_state_manager', 'saved_repositories_manager', 'schema_manager', + 'operation_queue', 'operation_history', 'cache_manager', +) +_SENTINEL = object() + + +class FakeResult: + """Stand-in for backup_manager.RestoreResult.""" + + def __init__(self, success=True, restored=None, errors=None, + plugins_to_install=None): + self.success = success + self.restored = restored if restored is not None else ["config"] + self.errors = errors or [] + self.plugins_to_install = plugins_to_install or [] + self.plugins_installed = [] + self.plugins_failed = [] + + def to_dict(self): + return { + "success": self.success, + "restored": self.restored, + "errors": self.errors, + "plugins_installed": self.plugins_installed, + "plugins_failed": self.plugins_failed, + } + + +@pytest.fixture +def client(): + originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS} + for name in _MANAGER_ATTRS: + setattr(api_v3, name, MagicMock()) + + app = Flask(__name__) + app.config["TESTING"] = True + app.register_blueprint(api_v3, url_prefix="/api/v3") + yield app.test_client() + + for name, original in originals.items(): + if original is _SENTINEL: + if hasattr(api_v3, name): + delattr(api_v3, name) + else: + setattr(api_v3, name, original) + + +@pytest.fixture +def restore(): + """Patch backup_manager.restore_backup (imported inside the handler).""" + with patch("src.backup_manager.restore_backup") as mock: + mock.return_value = FakeResult() + yield mock + + +def post(client, options=None, filename="backup.zip", content=b"PK\x03\x04fake"): + data = {"backup_file": (io.BytesIO(content), filename)} + if options is not None: + data["options"] = options + return client.post(URL, data=data, content_type="multipart/form-data") + + +class TestRequestValidation: + def test_missing_file_is_a_400(self, client, restore): + response = client.post(URL, data={}, content_type="multipart/form-data") + assert response.status_code == 400 + assert "No backup_file" in response.get_json()["message"] + restore.assert_not_called() + + def test_absent_options_defaults_to_a_full_restore(self, client, restore): + # Documented default, not the bug: omitting options entirely means + # "restore everything". + post(client) + options = restore.call_args[0][2] + assert options.restore_config is True + assert options.restore_secrets is True + assert options.reinstall_plugins is True + + def test_partial_options_are_honoured(self, client, restore): + post(client, options=json.dumps({ + "restore_secrets": False, "reinstall_plugins": False})) + options = restore.call_args[0][2] + assert options.restore_secrets is False + assert options.reinstall_plugins is False + assert options.restore_config is True # unspecified stays default + + @pytest.mark.parametrize("raw", ["{not json", "", "{'single': 'quotes'}"]) + def test_malformed_options_are_refused(self, client, restore, raw): + # Regression: this fell back to {}, and every flag defaults to + # True, so a caller asking for a narrow restore and mis-serializing + # it got a full one — secrets overwritten — and no warning. + response = post(client, options=raw) + assert response.status_code == 400 + assert "Invalid options" in response.get_json()["message"] + restore.assert_not_called() + + @pytest.mark.parametrize("raw", ["[1,2,3]", '"a string"', "42", "true", "null"]) + def test_options_that_are_not_an_object_are_refused(self, client, restore, raw): + response = post(client, options=raw) + assert response.status_code == 400 + restore.assert_not_called() + + def test_empty_object_is_accepted_as_all_defaults(self, client, restore): + assert post(client, options="{}").status_code == 200 + assert restore.call_args[0][2].restore_config is True + + +class TestSuccess: + def test_success_returns_the_result(self, client, restore): + restore.return_value = FakeResult(success=True, restored=["config", "secrets"]) + response = post(client) + assert response.status_code == 200 + body = response.get_json() + assert body["status"] == "success" + assert body["data"]["restored"] == ["config", "secrets"] + + def test_temp_file_is_cleaned_up(self, client, restore): + seen = {} + + def capture(path, project_root, options): + seen["path"] = Path(path) + assert seen["path"].exists() # present while restoring + return FakeResult() + + restore.side_effect = capture + post(client) + assert not seen["path"].exists() + + def test_temp_file_cleaned_up_even_when_restore_raises(self, client, restore): + seen = {} + + def blow_up(path, project_root, options): + seen["path"] = Path(path) + raise RuntimeError("corrupt archive") + + restore.side_effect = blow_up + response = post(client) + assert response.status_code == 500 + assert not seen["path"].exists() + + +class TestPluginReinstall: + def test_plugins_are_reinstalled_when_requested(self, client, restore): + restore.return_value = FakeResult( + plugins_to_install=[{"plugin_id": "clock"}, {"plugin_id": "weather"}]) + api_v3.plugin_store_manager.install_plugin.return_value = True + response = post(client) + assert response.status_code == 200 + assert response.get_json()["data"]["plugins_installed"] == ["clock", "weather"] + + def test_reinstall_skipped_when_not_requested(self, client, restore): + restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}]) + post(client, options=json.dumps({"reinstall_plugins": False})) + api_v3.plugin_store_manager.install_plugin.assert_not_called() + + def test_entries_without_a_plugin_id_are_skipped(self, client, restore): + restore.return_value = FakeResult(plugins_to_install=[{}, {"plugin_id": "clock"}]) + api_v3.plugin_store_manager.install_plugin.return_value = True + post(client) + assert api_v3.plugin_store_manager.install_plugin.call_count == 1 + + def test_failed_reinstall_turns_the_whole_restore_into_an_error( + self, client, restore): + # Pinned as intentional: file restoration succeeded and does not + # touch result.errors, but a user whose plugins did not come back + # should not be told the restore was a success. + restore.return_value = FakeResult( + success=True, plugins_to_install=[{"plugin_id": "clock"}]) + api_v3.plugin_store_manager.install_plugin.return_value = False + response = post(client) + assert response.status_code == 500 + body = response.get_json() + assert body["status"] == "error" + assert "clock" in body["message"] + + def test_message_names_what_landed_and_what_did_not(self, client, restore): + restore.return_value = FakeResult( + success=True, restored=["config", "fonts"], + plugins_to_install=[{"plugin_id": "clock"}]) + api_v3.plugin_store_manager.install_plugin.return_value = False + message = post(client).get_json()["message"] + assert "restored: config, fonts" in message + assert "plugins not reinstalled: clock" in message + + def test_install_exception_is_recorded_without_leaking_details( + self, client, restore): + restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}]) + api_v3.plugin_store_manager.install_plugin.side_effect = RuntimeError( + "/srv/internal/path exploded") + body = post(client).get_json() + failures = body["data"]["plugins_failed"] + assert failures[0]["plugin_id"] == "clock" + assert "/srv/internal/path" not in json.dumps(body) + + def test_missing_store_manager_is_reported_per_plugin(self, client, restore): + restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}]) + api_v3.plugin_store_manager = None + with patch("web_interface.blueprints.api_v3.plugin_store_manager", None): + body = post(client).get_json() + assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable" + + +class TestFailureReporting: + def test_restore_errors_produce_a_500(self, client, restore): + restore.return_value = FakeResult( + success=False, restored=[], errors=["config: permission denied"]) + response = post(client) + assert response.status_code == 500 + assert "permission denied" in response.get_json()["message"] + + def test_partial_restore_names_both_sides(self, client, restore): + restore.return_value = FakeResult( + success=False, restored=["config"], errors=["secrets: unwritable"]) + message = post(client).get_json()["message"] + assert "restored: config" in message + assert "failed: secrets: unwritable" in message + + def test_failure_without_detail_still_says_something(self, client, restore): + restore.return_value = FakeResult(success=False, restored=[], errors=[]) + message = post(client).get_json()["message"] + assert "Restore incomplete" in message + + def test_unexpected_exception_is_a_500(self, client, restore): + restore.side_effect = RuntimeError("boom") + response = post(client) + assert response.status_code == 500 + assert response.get_json()["status"] == "error" diff --git a/web_interface/blueprints/api_v3.py b/web_interface/blueprints/api_v3.py index e887e36f..e5f7d44c 100644 --- a/web_interface/blueprints/api_v3.py +++ b/web_interface/blueprints/api_v3.py @@ -8181,7 +8181,16 @@ def backup_restore(): try: opts_dict = json.loads(options_raw) except json.JSONDecodeError: - opts_dict = {} + opts_dict = None + if not isinstance(opts_dict, dict): + # Every option defaults to True, so falling back to {} on a + # parse failure would silently perform a FULL restore — + # secrets and all — for a caller who asked for a narrow one + # and mis-serialized it. Refuse instead of guessing. + return jsonify({ + 'status': 'error', + 'message': 'Invalid options: expected a JSON object', + }), 400 options = RestoreOptions( restore_config=bool(opts_dict.get('restore_config', True)), restore_secrets=bool(opts_dict.get('restore_secrets', True)),