From 2efd49a1ab763c0aa98d81529d8acd5d5cf3a82a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 17:33:47 +0000 Subject: [PATCH] refactor(web): use canonical secret helpers in api_v3; make ConfigManager secret strip/merge array-aware MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit api_v3.py carried three inline nested copies of find_secret_fields/ separate_secrets (main-config save, plugin-config save, plugin-config reset). They drifted from each other (one lacked isinstance guards) and none supported the canonical module's array-item secrets (accounts[].token). All three endpoints now import from src/web_interface/secret_helpers. Adopting the canonical behavior makes array-item secrets reachable, and their parallel-placeholder shape ([{'token': ...}, {}] alongside the regular list) was not survivable by ConfigManager's round-trip: _strip_secrets_recursive dropped the whole key (losing the regular fields from config.json) and _deep_merge replaced the regular list wholesale on load. Both are now array-aware: - strip removes the secret fields from each item and ALWAYS keeps the list so indices survive for merge-on-load; whole-key secrets (scalar lists, shape mismatches) still drop the key entirely — never leak. - merge folds each secrets item into the config item at the same index, skipping {} placeholders. The regular list's length is authoritative in both directions: a user deleting an array item never has it resurrected from a stale secrets entry (extras warn and are ignored). api_v3's own deep_merge intentionally still replaces lists wholesale — form posts carry complete arrays and index-merging would resurrect deleted items; a comment now documents that. Tests: the parity guard flips from 'exactly 3 inline copies' to 'zero, and the canonical import must exist'; TestArraySecretStripAndMerge covers the new strip/merge semantics incl. length-mismatch contracts; new test_api_v3_secret_roundtrip.py drives all three endpoints through a Flask client with a REAL ConfigManager+SchemaManager over tmp_path, proving secrets land in config_secrets.json, config.json stays clean, and a fresh load merges them back into the right array items. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh --- src/config_manager.py | 70 ++++- test/test_config_manager_secrets.py | 122 +++++++++ .../test_api_v3_secret_roundtrip.py | 259 ++++++++++++++++++ .../test_secret_separation_parity.py | 92 ++----- web_interface/blueprints/api_v3.py | 108 +------- 5 files changed, 478 insertions(+), 173 deletions(-) create mode 100644 test/web_interface/test_api_v3_secret_roundtrip.py diff --git a/src/config_manager.py b/src/config_manager.py index 569b295e..9872be6c 100644 --- a/src/config_manager.py +++ b/src/config_manager.py @@ -269,20 +269,47 @@ class ConfigManager: self.logger.error(error_msg, exc_info=True) raise ConfigError(error_msg, config_path=self.config_path) from e + @staticmethod + def _is_parallel_secrets_list(value: Any) -> bool: + """True for the parallel-placeholder list shape emitted by + ``secret_helpers.separate_secrets`` for array-item secrets: a + non-empty list whose elements are ALL dicts (``{}`` marks an item + with no secrets). Any other list-shaped secrets value is a + whole-key secret (e.g. a list of secret scalars).""" + return (isinstance(value, list) and bool(value) + and all(isinstance(item, dict) for item in value)) + def _strip_secrets_recursive(self, data_to_filter: Dict[str, Any], secrets: Dict[str, Any]) -> Dict[str, Any]: """Recursively remove secret keys from a dictionary.""" result = {} for key, value in data_to_filter.items(): - if key in secrets: - if isinstance(value, dict) and isinstance(secrets[key], dict): - # This key is a shared group, recurse - stripped_sub_dict = self._strip_secrets_recursive(value, secrets[key]) - if stripped_sub_dict: # Only add if there's non-secret data left - result[key] = stripped_sub_dict - # Else, it's a secret key at this level, so we skip it - else: + if key not in secrets: # This key is not in secrets, so we keep it result[key] = value + continue + sec = secrets[key] + if isinstance(value, dict) and isinstance(sec, dict): + # This key is a shared group, recurse + stripped_sub_dict = self._strip_secrets_recursive(value, sec) + if stripped_sub_dict: # Only add if there's non-secret data left + result[key] = stripped_sub_dict + elif isinstance(value, list) and self._is_parallel_secrets_list(sec): + # Parallel-list shape from separate_secrets: sec[i] holds the + # secret fields of value[i] ({} = item i has none). Strip each + # item and ALWAYS keep the list — indices must survive so the + # merge-on-load can realign secrets with their items. The + # regular list's length is authoritative: extra secrets + # entries are ignored. + stripped_items = [] + for i, item in enumerate(value): + s_item = sec[i] if i < len(sec) else {} + if isinstance(item, dict) and s_item: + stripped_items.append(self._strip_secrets_recursive(item, s_item)) + else: + stripped_items.append(item) + result[key] = stripped_items + # Else: whole-key secret (scalar, list of secret scalars, or a + # shape mismatch) -> drop the key entirely. Never leak. return result def _load_secrets_for_save(self) -> Dict[str, Any]: @@ -358,11 +385,36 @@ class ConfigManager: return None def _deep_merge(self, target: Dict[str, Any], source: Dict[str, Any]) -> None: - """Deep merge source dict into target dict.""" + """Deep merge source dict into target dict. + + Sole call site: merging config_secrets.json into the loaded config. + Understands the parallel-list shape separate_secrets emits for + array-item secrets (see _is_parallel_secrets_list): each secrets + list item is merged into the config list item at the same index + ({} placeholders skipped). The config list's length is + authoritative — a user deleting an array item from config.json + must not have it resurrected from a stale secrets entry.""" for key, value in source.items(): if key in target and isinstance(target[key], dict) and isinstance(value, dict): self._deep_merge(target[key], value) + elif (key in target and isinstance(target[key], list) + and self._is_parallel_secrets_list(value)): + tlist = target[key] + for i, s_item in enumerate(value): + if i >= len(tlist): + self.logger.warning( + "Secrets list for %r is longer than the config list " + "(%d > %d); ignoring the extra entries", + key, len(value), len(tlist)) + break + if not s_item: + continue # {} placeholder: item i has no secrets + if isinstance(tlist[i], dict): + self._deep_merge(tlist[i], s_item) + else: + tlist[i] = s_item # shape drift; the secret wins else: + # Scalars AND whole-secret scalar arrays: replace (legacy). target[key] = value def _create_config_from_template(self) -> None: diff --git a/test/test_config_manager_secrets.py b/test/test_config_manager_secrets.py index 94c7a6a2..83452bf0 100644 --- a/test/test_config_manager_secrets.py +++ b/test/test_config_manager_secrets.py @@ -177,3 +177,125 @@ class TestLoadFastPath: os.utime(config_file, ns=(1_000_000_000, 1_000_000_000)) assert manager.load_config()["timezone"] == "AAA" # stale, by design + + +class TestArraySecretStripAndMerge: + """Array-item secrets round-trip (parallel-placeholder lists). + + secret_helpers.separate_secrets emits array secrets as a list parallel + to the regular list, with {} for items that carry no secrets. Strip + must remove the secret fields from config.json while preserving item + indices; load must merge them back into the right items. The regular + list's length is authoritative in both directions. + """ + + def test_strip_removes_array_item_secrets_keeps_indices(self, tmp_path): + manager = make_manager(tmp_path) + data = {"plugin": {"accounts": [ + {"name": "a", "token": "ta"}, + {"name": "b"}, + ]}} + secrets = {"plugin": {"accounts": [{"token": "ta"}, {}]}} + stripped = manager._strip_secrets_recursive(data, secrets) + assert stripped == {"plugin": {"accounts": [{"name": "a"}, {"name": "b"}]}} + + def test_strip_keeps_all_placeholder_items(self, tmp_path): + # Even when every item strips to nothing extra, the list survives + # with its indices — required for merge-on-load alignment. + manager = make_manager(tmp_path) + data = {"accounts": [{"token": "t1"}, {"token": "t2"}]} + secrets = {"accounts": [{"token": "t1"}, {"token": "t2"}]} + stripped = manager._strip_secrets_recursive(data, secrets) + assert stripped == {"accounts": [{}, {}]} + + def test_strip_whole_scalar_array_secret_drops_key(self, tmp_path): + # A list of secret scalars is a whole-key secret, not the parallel + # shape — the key must vanish from config.json entirely. + manager = make_manager(tmp_path) + data = {"recovery_codes": ["a", "b"], "city": "Austin"} + secrets = {"recovery_codes": ["a", "b"]} + stripped = manager._strip_secrets_recursive(data, secrets) + assert stripped == {"city": "Austin"} + + def test_strip_shape_mismatch_drops_key(self, tmp_path): + # Conservative contract: if the shapes disagree, never leak. + manager = make_manager(tmp_path) + data = {"accounts": {"name": "not-a-list"}} + secrets = {"accounts": [{"token": "t"}]} + stripped = manager._strip_secrets_recursive(data, secrets) + assert stripped == {} + + def test_strip_ignores_extra_secrets_entries(self, tmp_path): + # Regular list length is authoritative: a user deleted an item. + manager = make_manager(tmp_path) + data = {"accounts": [{"name": "a", "token": "ta"}]} + secrets = {"accounts": [{"token": "ta"}, {"token": "tb"}]} + stripped = manager._strip_secrets_recursive(data, secrets) + assert stripped == {"accounts": [{"name": "a"}]} + + def test_merge_restores_array_item_secrets(self, tmp_path): + manager = make_manager(tmp_path) + target = {"accounts": [{"name": "a"}, {"name": "b"}]} + manager._deep_merge(target, {"accounts": [{"token": "ta"}, {}]}) + assert target == {"accounts": [ + {"name": "a", "token": "ta"}, + {"name": "b"}, + ]} + + def test_merge_ignores_extra_secrets_entries_with_warning(self, tmp_path, caplog): + manager = make_manager(tmp_path) + target = {"accounts": [{"name": "a"}]} + with caplog.at_level("WARNING"): + manager._deep_merge( + target, {"accounts": [{"token": "ta"}, {"token": "ghost"}]}) + assert target == {"accounts": [{"name": "a", "token": "ta"}]} + assert any("longer than the config list" in r.message for r in caplog.records) + + def test_merge_non_dict_item_replaced_by_secret(self, tmp_path): + # Shape drift inside the list: the secret wins for that index. + manager = make_manager(tmp_path) + target = {"accounts": ["oddball", {"name": "b"}]} + manager._deep_merge(target, {"accounts": [{"token": "ta"}, {}]}) + assert target == {"accounts": [{"token": "ta"}, {"name": "b"}]} + + def test_merge_whole_scalar_array_still_replaces(self, tmp_path): + # Legacy behavior preserved: a non-parallel list replaces wholesale. + manager = make_manager(tmp_path) + target = {"recovery_codes": ["old"]} + manager._deep_merge(target, {"recovery_codes": ["new1", "new2"]}) + assert target == {"recovery_codes": ["new1", "new2"]} + + def test_full_save_load_round_trip(self, tmp_path): + # End to end on real files: save strips array secrets out of + # config.json; load merges them back into the right items. + manager = make_manager( + tmp_path, + config={"plugin": {"accounts": [ + {"name": "a", "token": "s3cret-a"}, + {"name": "b", "token": "s3cret-b"}, + ]}}, + secrets={"plugin": {"accounts": [ + {"token": "s3cret-a"}, {"token": "s3cret-b"}, + ]}}, + ) + loaded = manager.load_config() + assert loaded["plugin"]["accounts"][0]["token"] == "s3cret-a" + + manager.save_config(loaded) + + raw = (tmp_path / "config.json").read_text() + assert "s3cret" not in raw + on_disk = json.loads(raw) + assert on_disk["plugin"]["accounts"] == [{"name": "a"}, {"name": "b"}] + + # A fresh manager (constructed directly — make_manager would + # overwrite the just-saved config.json) re-merges from the secrets + # file on load. + fresh = ConfigManager(config_path=str(tmp_path / "config.json"), + secrets_path=str(tmp_path / "config_secrets.json")) + fresh.template_path = str(tmp_path / "no-template.json") + reloaded = fresh.load_config() + assert reloaded["plugin"]["accounts"] == [ + {"name": "a", "token": "s3cret-a"}, + {"name": "b", "token": "s3cret-b"}, + ] diff --git a/test/web_interface/test_api_v3_secret_roundtrip.py b/test/web_interface/test_api_v3_secret_roundtrip.py new file mode 100644 index 00000000..a4648824 --- /dev/null +++ b/test/web_interface/test_api_v3_secret_roundtrip.py @@ -0,0 +1,259 @@ +""" +End-to-end secret round-trips through the three api_v3 endpoints that +separate secrets from regular config (main-config save, plugin-config save, +plugin-config reset) — now backed by the canonical +src/web_interface/secret_helpers implementations. + +Unlike test_web_api.py (which mocks the config manager), these tests run a +REAL ConfigManager and a REAL SchemaManager over tmp_path files, so they +prove the whole chain: endpoint separation -> config_secrets.json write -> +atomic config.json save (strip) -> load_config (merge back), including the +array-item secret shape (accounts[].token) the inline copies never +supported. +""" + +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 src.config_manager import ConfigManager # noqa: E402 +from src.plugin_system.schema_manager import SchemaManager # noqa: E402 +from web_interface.blueprints.api_v3 import api_v3 # noqa: E402 + + +PLUGIN_ID = "testplugin" + +SCHEMA = { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "enabled": {"type": "boolean", "default": True}, + "display_duration": {"type": "number", "default": 15}, + "api_key": {"type": "string", "x-secret": True, "default": ""}, + "city": {"type": "string", "default": "Austin"}, + "accounts": { + "type": "array", + "default": [], + "items": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "token": {"type": "string", "x-secret": True}, + }, + }, + }, + }, +} + + +@pytest.fixture +def env(tmp_path): + """Real ConfigManager + SchemaManager over tmp_path, wired onto the + api_v3 blueprint with the remaining managers mocked.""" + config_file = tmp_path / "config.json" + config_file.write_text("{}") + + plugins_dir = tmp_path / "plugins" + plugin_dir = plugins_dir / PLUGIN_ID + plugin_dir.mkdir(parents=True) + (plugin_dir / "config_schema.json").write_text(json.dumps(SCHEMA)) + (plugin_dir / "manifest.json").write_text(json.dumps({ + "id": PLUGIN_ID, "name": "Test Plugin", "version": "1.0.0", + })) + + config_manager = ConfigManager( + config_path=str(config_file), + secrets_path=str(tmp_path / "config_secrets.json")) + config_manager.template_path = str(tmp_path / "no-template.json") + + schema_manager = SchemaManager(plugins_dir=plugins_dir, + project_root=tmp_path) + + plugin_manager = MagicMock() + plugin_manager.plugin_manifests = {PLUGIN_ID: {"id": PLUGIN_ID}} + plugin_manager.plugins_dir = plugins_dir + plugin_manager.get_plugin.return_value = None + + api_v3.config_manager = config_manager + api_v3.schema_manager = schema_manager + api_v3.plugin_manager = plugin_manager + api_v3.plugin_store_manager = MagicMock() + api_v3.saved_repositories_manager = MagicMock() + api_v3.operation_queue = MagicMock() + api_v3.plugin_state_manager = MagicMock() + api_v3.operation_history = MagicMock() + api_v3.cache_manager = 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.config_manager = config_manager + e.config_file = config_file + e.secrets_file = tmp_path / "config_secrets.json" + e.tmp_path = tmp_path + + def fresh_load(): + """Load via a NEW ConfigManager, as the next request/process would. + + The endpoint's manager serves its post-save in-memory config via the + mtime fast path, and that copy predates the secrets it just + separated out — a pre-existing quirk that applies to scalar secrets + too. On-disk truth is what these tests care about. + """ + fresh = ConfigManager(config_path=str(config_file), + secrets_path=str(e.secrets_file)) + fresh.template_path = str(tmp_path / "no-template.json") + return fresh.load_config() + + e.fresh_load = fresh_load + return e + + +def _on_disk(path): + return json.loads(path.read_text()) + + +class TestSaveMainConfig: + """Site A: POST /config/main with a plugin-id key.""" + + def test_array_and_scalar_secrets_routed_to_secrets_file(self, env): + resp = env.client.post("/api/v3/config/main", json={ + PLUGIN_ID: { + "city": "Dallas", + "api_key": "s3cret-key", + "accounts": [ + {"name": "a", "token": "s3cret-a"}, + {"name": "b"}, + ], + }, + }) + assert resp.status_code == 200, resp.get_json() + + on_disk = _on_disk(env.config_file) + assert on_disk[PLUGIN_ID]["city"] == "Dallas" + assert "api_key" not in on_disk[PLUGIN_ID] + assert on_disk[PLUGIN_ID]["accounts"] == [{"name": "a"}, {"name": "b"}] + assert "s3cret" not in env.config_file.read_text() + + secrets = _on_disk(env.secrets_file) + assert secrets[PLUGIN_ID]["api_key"] == "s3cret-key" + assert secrets[PLUGIN_ID]["accounts"] == [{"token": "s3cret-a"}, {}] + + def test_load_config_merges_secrets_back(self, env): + env.client.post("/api/v3/config/main", json={ + PLUGIN_ID: {"accounts": [{"name": "a", "token": "s3cret-a"}]}, + }) + merged = env.fresh_load() + assert merged[PLUGIN_ID]["accounts"] == [ + {"name": "a", "token": "s3cret-a"}] + + +class TestSavePluginConfig: + """Site B: POST /plugins/config (JSON body).""" + + def _save(self, env, config): + return env.client.post("/api/v3/plugins/config", json={ + "plugin_id": PLUGIN_ID, "config": config, + }) + + def test_round_trip_with_array_secrets(self, env): + resp = self._save(env, { + "enabled": True, + "city": "Houston", + "api_key": "s3cret-key", + "accounts": [ + {"name": "a", "token": "s3cret-a"}, + {"name": "b", "token": "s3cret-b"}, + ], + }) + assert resp.status_code == 200, resp.get_json() + + assert "s3cret" not in env.config_file.read_text() + on_disk = _on_disk(env.config_file) + assert on_disk[PLUGIN_ID]["accounts"] == [{"name": "a"}, {"name": "b"}] + + secrets = _on_disk(env.secrets_file) + assert secrets[PLUGIN_ID]["accounts"] == [ + {"token": "s3cret-a"}, {"token": "s3cret-b"}] + + merged = env.fresh_load() + assert merged[PLUGIN_ID]["accounts"][1]["token"] == "s3cret-b" + + def test_secret_count_message_counts_top_level_keys(self, env): + # Pinned: the "(N secret field(s))" message counts TOP-LEVEL keys of + # the separated secrets dict. Here that is 2: the posted accounts + # array (all its item tokens count as ONE key) plus the schema's + # api_key default ("") that merge_with_defaults adds before + # separation. + resp = self._save(env, { + "accounts": [{"name": "a", "token": "t"}], + }) + message = resp.get_json()["message"] + assert "(2 secret field(s) saved to config_secrets.json)" in message + + def test_resave_replaces_stored_secrets_list_wholesale(self, env): + # Characterized: api_v3's deep_merge intentionally replaces lists, + # so a re-save's parallel secrets list is authoritative. + self._save(env, {"accounts": [ + {"name": "a", "token": "old-a"}, + {"name": "b", "token": "old-b"}, + ]}) + self._save(env, {"accounts": [{"name": "only", "token": "new-only"}]}) + + secrets = _on_disk(env.secrets_file) + assert secrets[PLUGIN_ID]["accounts"] == [{"token": "new-only"}] + merged = env.fresh_load() + assert merged[PLUGIN_ID]["accounts"] == [ + {"name": "only", "token": "new-only"}] + + +class TestResetPluginConfig: + """Site C: POST /plugins/config/reset.""" + + def _seed(self, env): + env.client.post("/api/v3/plugins/config", json={ + "plugin_id": PLUGIN_ID, + "config": {"city": "Houston", "api_key": "s3cret-key", + "accounts": [{"name": "a", "token": "s3cret-a"}]}, + }) + + def test_reset_preserving_secrets(self, env): + self._seed(env) + resp = env.client.post("/api/v3/plugins/config/reset", json={ + "plugin_id": PLUGIN_ID, "preserve_secrets": True, + }) + assert resp.status_code == 200, resp.get_json() + + on_disk = _on_disk(env.config_file) + assert on_disk[PLUGIN_ID]["city"] == "Austin" # schema default + assert on_disk[PLUGIN_ID]["accounts"] == [] # schema default + + # Existing secrets survive (top-level-only preserve merge, pinned). + secrets = _on_disk(env.secrets_file) + assert secrets[PLUGIN_ID]["api_key"] == "s3cret-key" + assert secrets[PLUGIN_ID]["accounts"] == [{"token": "s3cret-a"}] + + def test_reset_without_preserving_secrets(self, env): + self._seed(env) + resp = env.client.post("/api/v3/plugins/config/reset", json={ + "plugin_id": PLUGIN_ID, "preserve_secrets": False, + }) + assert resp.status_code == 200, resp.get_json() + + secrets = _on_disk(env.secrets_file) + # Replaced with schema-default secrets — the schema declares no + # secret defaults, so the plugin's secrets are emptied. + assert secrets[PLUGIN_ID] in ({}, {"api_key": ""}) diff --git a/test/web_interface/test_secret_separation_parity.py b/test/web_interface/test_secret_separation_parity.py index 65c5576e..91367234 100644 --- a/test/web_interface/test_secret_separation_parity.py +++ b/test/web_interface/test_secret_separation_parity.py @@ -1,18 +1,13 @@ """ -Drift guard for the duplicated secret-separation logic. +Drift guard: api_v3 must use the canonical secret helpers. -src/web_interface/secret_helpers.py is the canonical implementation of -find_secret_fields / separate_secrets, but web_interface/blueprints/api_v3.py -still carries THREE inline nested-function copies of each (in the plugin -config GET, POST, and reset endpoints). The copies lack the canonical -module's array-item support (`accounts[].token`), so migrating an endpoint -onto the module is a behavior change that must be made deliberately. - -This file guards two things: -1. The copy count can only go DOWN. A fourth copy appearing means someone - re-implemented the logic again instead of importing secret_helpers. -2. The known behavioral gap is documented as an executable fact, so whoever - migrates the endpoints knows exactly what changes. +Historically web_interface/blueprints/api_v3.py carried THREE inline +nested-function copies of ``find_secret_fields``/``separate_secrets`` (in the +main-config save, plugin-config save, and plugin-config reset endpoints). +They lacked the canonical module's array-item secret support and drifted from +each other. They have been migrated onto +``src/web_interface/secret_helpers`` — this file now guards against copies +REAPPEARING, and keeps the canonical array-item behavior executable. """ import re @@ -23,77 +18,44 @@ from src.web_interface.secret_helpers import find_secret_fields, separate_secret API_V3_PATH = (Path(__file__).resolve().parents[2] / "web_interface" / "blueprints" / "api_v3.py") -# Update DOWNWARD as endpoints migrate onto src/web_interface/secret_helpers. -EXPECTED_INLINE_COPIES = 3 +# The migration is complete: any inline reimplementation is a regression. +EXPECTED_INLINE_COPIES = 0 -class TestInlineCopyCount: +class TestNoInlineCopies: def _count(self, name: str) -> int: source = API_V3_PATH.read_text(encoding="utf-8") return len(re.findall(rf"^\s*def {name}\(", source, flags=re.MULTILINE)) - def test_find_secret_fields_copy_count(self): + def test_no_inline_find_secret_fields(self): count = self._count("find_secret_fields") assert count == EXPECTED_INLINE_COPIES, ( f"api_v3.py has {count} inline find_secret_fields definitions, " - f"expected {EXPECTED_INLINE_COPIES}. New code must import it from " - f"src/web_interface/secret_helpers instead of re-implementing it; " - f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES." + f"expected {EXPECTED_INLINE_COPIES}. Import it from " + f"src/web_interface/secret_helpers instead of re-implementing it." ) - def test_separate_secrets_copy_count(self): + def test_no_inline_separate_secrets(self): count = self._count("separate_secrets") assert count == EXPECTED_INLINE_COPIES, ( f"api_v3.py has {count} inline separate_secrets definitions, " - f"expected {EXPECTED_INLINE_COPIES}. New code must import it from " - f"src/web_interface/secret_helpers instead of re-implementing it; " - f"if you migrated an endpoint, lower EXPECTED_INLINE_COPIES." + f"expected {EXPECTED_INLINE_COPIES}. Import it from " + f"src/web_interface/secret_helpers instead of re-implementing it." ) - def test_inline_copies_lack_array_item_support(self): - """The documented gap: no inline copy recurses into array `items` - schemas, so array-item secrets (accounts[].token) are NOT routed to - config_secrets.json by these endpoints. The canonical module handles - them. When an endpoint migrates onto the module that behavior - changes (a fix, but a deliberate one). - - If this fails, an inline copy has grown array support — duplicating - the canonical module even harder. Migrate the endpoint onto - src/web_interface/secret_helpers instead. - """ - for body in self._inline_bodies("find_secret_fields"): - # Array handling requires checking type == 'array'; no inline - # copy does. (Can't grep bare "items" — properties.items() the - # dict method appears legitimately.) - assert "'array'" not in body and '"array"' not in body - - @staticmethod - def _inline_bodies(name: str): - """Extract each inline def's body from api_v3.py by indentation.""" - lines = API_V3_PATH.read_text(encoding="utf-8").splitlines() - bodies = [] - i = 0 - while i < len(lines): - match = re.match(rf"^(\s+)def {name}\(", lines[i]) - if not match: - i += 1 - continue - indent = len(match.group(1)) - body = [lines[i]] - i += 1 - while i < len(lines): - line = lines[i] - if line.strip() and (len(line) - len(line.lstrip())) <= indent: - break - body.append(line) - i += 1 - bodies.append("\n".join(body)) - assert bodies, f"no inline {name} definitions found" - return bodies + def test_canonical_import_present(self): + # Tripwire: the endpoints still need the helpers, so removing the + # import means either dead secret handling or a new local copy. + source = API_V3_PATH.read_text(encoding="utf-8") + assert re.search( + r"from src\.web_interface\.secret_helpers import .*find_secret_fields", + source, + ), "api_v3.py no longer imports the canonical secret helpers" class TestCanonicalArrayItemBehavior: - """Executable documentation of what migrating endpoints will change.""" + """Executable documentation of the array-item secret contract the + endpoints now inherit from the canonical module.""" SCHEMA = { "accounts": { diff --git a/web_interface/blueprints/api_v3.py b/web_interface/blueprints/api_v3.py index 56f91a8b..2a0a91cf 100644 --- a/web_interface/blueprints/api_v3.py +++ b/web_interface/blueprints/api_v3.py @@ -21,6 +21,7 @@ logger = logging.getLogger(__name__) # Import new infrastructure from src.web_interface.api_helpers import success_response, error_response, validate_request_json from src.web_interface.errors import ErrorCode +from src.web_interface.secret_helpers import find_secret_fields, separate_secrets from src.plugin_system.operation_types import OperationType from src.web_interface.validators import ( validate_file_upload @@ -1196,18 +1197,6 @@ def save_main_config(): plugins_dir = PROJECT_ROOT / plugins_dir_name schema_path = plugins_dir / plugin_id / 'config_schema.json' - def find_secret_fields(properties, prefix=''): - """Recursively find fields marked with x-secret: true""" - fields = set() - for field_name, field_props in properties.items(): - full_path = f"{prefix}.{field_name}" if prefix else field_name - if field_props.get('x-secret', False): - fields.add(full_path) - # Check nested objects - if field_props.get('type') == 'object' and 'properties' in field_props: - fields.update(find_secret_fields(field_props['properties'], full_path)) - return fields - if schema_path.exists(): try: with open(schema_path, 'r', encoding='utf-8') as f: @@ -1218,24 +1207,6 @@ def save_main_config(): logger.debug("Error reading schema for secret detection: %s", e) # Separate secrets from regular config (same logic as save_plugin_config) - def separate_secrets(config, secrets_set, prefix=''): - """Recursively separate secret fields from regular config""" - regular = {} - secrets = {} - for key, value in config.items(): - full_path = f"{prefix}.{key}" if prefix else key - if isinstance(value, dict): - nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path) - if nested_regular: - regular[key] = nested_regular - if nested_secrets: - secrets[key] = nested_secrets - elif full_path in secrets_set: - secrets[key] = value - else: - regular[key] = value - return regular, secrets - regular_config, secrets_config = separate_secrets(plugin_config, secret_fields) # PRE-PROCESSING: Preserve 'enabled' state if not in regular_config @@ -4039,6 +4010,11 @@ def deep_merge(base_dict, update_dict): """ Deep merge update_dict into base_dict. For nested dicts, recursively merge. For other types, update_dict takes precedence. + + Lists are intentionally REPLACED wholesale, never index-merged: form posts + carry complete arrays, and index-merging would resurrect items the user + deleted. This also applies to the parallel secrets lists produced by + separate_secrets — a newly saved secrets list is authoritative. """ result = base_dict.copy() for key, value in update_dict.items(): @@ -5053,23 +5029,9 @@ def save_plugin_config(): # Default to True on error to avoid disabling plugins plugin_config['enabled'] = True - # Find secret fields (supports nested schemas) + # Find secret fields (supports nested schemas and array-item secrets) secret_fields = set() - def find_secret_fields(properties, prefix=''): - """Recursively find fields marked with x-secret: true""" - fields = set() - if not isinstance(properties, dict): - return fields - for field_name, field_props in properties.items(): - full_path = f"{prefix}.{field_name}" if prefix else field_name - if isinstance(field_props, dict) and field_props.get('x-secret', False): - fields.add(full_path) - # Check nested objects - if isinstance(field_props, dict) and field_props.get('type') == 'object' and 'properties' in field_props: - fields.update(find_secret_fields(field_props['properties'], full_path)) - return fields - if schema and 'properties' in schema: secret_fields = find_secret_fields(schema['properties']) @@ -5384,29 +5346,8 @@ def save_plugin_config(): status_code=400 ) - # Separate secrets from regular config (handles nested configs) - def separate_secrets(config, secrets_set, prefix=''): - """Recursively separate secret fields from regular config""" - regular = {} - secrets = {} - - for key, value in config.items(): - full_path = f"{prefix}.{key}" if prefix else key - - if isinstance(value, dict): - # Recursively handle nested dicts - nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path) - if nested_regular: - regular[key] = nested_regular - if nested_secrets: - secrets[key] = nested_secrets - elif full_path in secrets_set: - secrets[key] = value - else: - regular[key] = value - - return regular, secrets - + # Separate secrets from regular config (handles nested configs and + # array-item secrets — see src/web_interface/secret_helpers.py) regular_config, secrets_config = separate_secrets(plugin_config, secret_fields) # Get current configs @@ -5654,41 +5595,10 @@ def reset_plugin_config(): schema = schema_mgr.load_schema(plugin_id, use_cache=True) secret_fields = set() - def find_secret_fields(properties, prefix=''): - """Recursively find fields marked with x-secret: true""" - fields = set() - if not isinstance(properties, dict): - return fields - for field_name, field_props in properties.items(): - full_path = f"{prefix}.{field_name}" if prefix else field_name - if isinstance(field_props, dict) and field_props.get('x-secret', False): - fields.add(full_path) - if isinstance(field_props, dict) and field_props.get('type') == 'object' and 'properties' in field_props: - fields.update(find_secret_fields(field_props['properties'], full_path)) - return fields - if schema and 'properties' in schema: secret_fields = find_secret_fields(schema['properties']) # Separate defaults into regular and secret configs - def separate_secrets(config, secrets_set, prefix=''): - """Recursively separate secret fields from regular config""" - regular = {} - secrets = {} - for key, value in config.items(): - full_path = f"{prefix}.{key}" if prefix else key - if isinstance(value, dict): - nested_regular, nested_secrets = separate_secrets(value, secrets_set, full_path) - if nested_regular: - regular[key] = nested_regular - if nested_secrets: - secrets[key] = nested_secrets - elif full_path in secrets_set: - secrets[key] = value - else: - regular[key] = value - return regular, secrets - default_regular, default_secrets = separate_secrets(defaults, secret_fields) # Update main config with defaults