gabriel/muse — blame/sha256:2/test_cmd_config_hardening.py

1 files

1 commits

0 hotspots

0 🧊 dead

0 💥 blast risk

sha256:4 Merge branch 'dev' into main · gabriel · Jun 17, 2026

1	"""Comprehensive hardening tests for ``muse config``.
2
3	Coverage
4	--------
5	Unit — muse/cli/config.py internals
6	- _escape: backslash, quote, newline, carriage-return, null-byte escaping
7	- _validate_toml_key: unsafe characters rejected, safe keys pass
8	- _dump_toml: limits section round-trips, domain key injection blocked,
9	remote name injection blocked; [user] section is never emitted
10	- set_config_value: limits namespace writes correctly, TOML key injection
11	blocked, unknown limits key rejected, non-integer limits rejected,
12	invalid shard_prefix_length rejected, user.* namespace raises ValueError
13	- get_config_value: limits keys read back correctly after write
14	- config_as_dict: limits section included in output; [user] section absent
15
16	Integration — CLI commands via CliRunner
17	- run_show: TOML text and JSON outputs, limits displayed in both formats,
18	sanitize_display applied in text mode, --format json alias
19	- run_get: bare value to stdout, --json schema, not-set exits nonzero,
20	key sanitized in stderr message
21	- run_set: success to stderr, --json schema, blocked namespace rejected,
22	TOML injection rejected, limits set and readable, non-integer limits rejected,
23	user.* namespace blocked (identity lives in identity.toml)
24	- run_edit: no-repo exits, missing config exits, bad editor exits
25
26	Security
27	- TOML key injection: newline, bracket, equals, quote in domain key blocked
28	- ANSI in key/value sanitized in run_show text mode
29	- ANSI in exception message sanitized in run_set stderr
30	- run_set success message goes to stderr (stdout clean for scripting)
31	- run_get error goes to stderr
32
33	E2E (full round-trip via CLI)
34	- set then get is consistent for hub/domain/limits keys
35	- set limits then show --json contains limits
36	- set domain then show TOML is valid TOML
37	- limits fall-through to domain is fixed (writes to [limits] not [domain])
38	- user.* set is blocked end-to-end
39
40	Stress
41	- 8 concurrent set_config_value calls to isolated repos: no corruption
42	"""
43
44	from __future__ import annotations
45
46	import json
47	import pathlib
48	import threading
49	import tomllib
50	from typing import TYPE_CHECKING
51	from unittest.mock import MagicMock, patch
52
53	import pytest
54
55	from muse.core.paths import config_toml_path, muse_dir
56
57	from tests.cli_test_helper import CliRunner, InvokeResult
58
59	if TYPE_CHECKING:
60	pass
61
62	from muse.cli.commands.config_cmd import _GetJson, _SetJson
63	from muse.core.types import JsonValue
64
65	type _ReadJson = dict[str, JsonValue]
66	from muse.core.types import MsgpackDict
67
68	cli = None
69	runner = CliRunner()
70
71	# ── fixtures ──────────────────────────────────────────────────────────────────
72
73
74	@pytest.fixture
75	def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
76	"""Minimal .muse/ repo with an empty config.toml."""
77	from muse._version import __version__
78
79	dot_muse = muse_dir(tmp_path)
80	for sub in ("refs/heads", "objects", "commits", "snapshots"):
81	(dot_muse / sub).mkdir(parents=True, exist_ok=True)
82	(dot_muse / "repo.json").write_text(
83	json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
84	)
85	(dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
86	(dot_muse / "refs" / "heads" / "main").write_text("")
87	(dot_muse / "config.toml").write_text("")
88	monkeypatch.chdir(tmp_path)
89	return tmp_path
90
91
92	def _json_get(result: InvokeResult) -> _GetJson:
93	for line in result.output.splitlines():
94	stripped = line.strip()
95	if stripped.startswith("{"):
96	d: _GetJson = json.loads(stripped)
97	return d
98	raise ValueError(f"No JSON in output:\n{result.output!r}")
99
100
101	def _json_set(result: InvokeResult) -> _SetJson:
102	for line in result.output.splitlines():
103	stripped = line.strip()
104	if stripped.startswith("{"):
105	d: _SetJson = json.loads(stripped)
106	return d
107	raise ValueError(f"No JSON in output:\n{result.output!r}")
108
109
110	def _json_show(result: InvokeResult) -> _ReadJson:
111	"""Extract the config payload from ``muse config read --json`` output.
112
113	The envelope wraps config sections under a ``config`` key; return that
114	inner dict so callers can access ``data["hub"]`` / ``data["domain"]`` etc.
115	directly without caring about the envelope structure.
116	"""
117	lines = result.output.splitlines()
118	start = None
119	for i, line in enumerate(lines):
120	if line.strip().startswith("{"):
121	start = i
122	break
123	if start is None:
124	raise ValueError(f"No JSON in output:\n{result.output!r}")
125	depth = 0
126	collected: list[str] = []
127	for line in lines[start:]:
128	collected.append(line)
129	depth += line.count("{") - line.count("}")
130	if depth <= 0:
131	break
132	envelope = json.loads("\n".join(collected))
133	# Unwrap envelope: return the inner config dict when present so assertions
134	# like data["hub"] work without knowing the envelope structure.
135	return envelope.get("config", envelope) # type: ignore[return-value]
136
137
138	# ── Unit: _escape ─────────────────────────────────────────────────────────────
139
140
141	class TestEscape:
142	def test_backslash_escaped(self) -> None:
143	from muse.cli.config import _escape
144	assert _escape("back\\slash") == "back\\\\slash"
145
146	def test_double_quote_escaped(self) -> None:
147	from muse.cli.config import _escape
148	assert _escape('say "hi"') == 'say \\"hi\\"'
149
150	def test_newline_escaped(self) -> None:
151	from muse.cli.config import _escape
152	assert _escape("line1\nline2") == "line1\\nline2"
153
154	def test_carriage_return_escaped(self) -> None:
155	from muse.cli.config import _escape
156	assert _escape("text\rmore") == "text\\rmore"
157
158	def test_null_byte_removed(self) -> None:
159	from muse.cli.config import _escape
160	assert "\0" not in _escape("bad\0byte")
161
162	def test_clean_string_passthrough(self) -> None:
163	from muse.cli.config import _escape
164	assert _escape("hello world") == "hello world"
165
166
167	# ── Unit: _validate_toml_key ──────────────────────────────────────────────────
168
169
170	class TestValidateTomlKey:
171	def test_newline_in_key_rejected(self) -> None:
172	from muse.cli.config import _validate_toml_key
173	with pytest.raises(ValueError, match="TOML keys"):
174	_validate_toml_key("bad\nkey")
175
176	def test_carriage_return_rejected(self) -> None:
177	from muse.cli.config import _validate_toml_key
178	with pytest.raises(ValueError, match="TOML keys"):
179	_validate_toml_key("bad\rkey")
180
181	def test_closing_bracket_rejected(self) -> None:
182	from muse.cli.config import _validate_toml_key
183	with pytest.raises(ValueError, match="TOML keys"):
184	_validate_toml_key("x]injection")
185
186	def test_opening_bracket_rejected(self) -> None:
187	from muse.cli.config import _validate_toml_key
188	with pytest.raises(ValueError, match="TOML keys"):
189	_validate_toml_key("[malicious")
190
191	def test_equals_rejected(self) -> None:
192	from muse.cli.config import _validate_toml_key
193	with pytest.raises(ValueError, match="TOML keys"):
194	_validate_toml_key("k=v")
195
196	def test_double_quote_rejected(self) -> None:
197	from muse.cli.config import _validate_toml_key
198	with pytest.raises(ValueError, match="TOML keys"):
199	_validate_toml_key('key"val')
200
201	def test_null_byte_rejected(self) -> None:
202	from muse.cli.config import _validate_toml_key
203	with pytest.raises(ValueError, match="TOML keys"):
204	_validate_toml_key("bad\0key")
205
206	def test_safe_key_passes(self) -> None:
207	from muse.cli.config import _validate_toml_key
208	_validate_toml_key("ticks_per_beat")
209	_validate_toml_key("my-key.123")
210	_validate_toml_key("CamelCase")
211
212
213	# ── Unit: _dump_toml ──────────────────────────────────────────────────────────
214
215
216	class TestDumpToml:
217	def test_limits_section_round_trips(self) -> None:
218	from muse.cli.config import LimitsConfig, MuseConfig, _dump_toml
219	cfg: MuseConfig = {"limits": LimitsConfig(max_walk_commits=99, max_ancestors=500)}
220	toml_text = _dump_toml(cfg)
221	parsed = tomllib.loads(toml_text)
222	assert parsed["limits"]["max_walk_commits"] == 99
223	assert parsed["limits"]["max_ancestors"] == 500
224
225	def test_limits_shard_prefix_length_written(self) -> None:
226	from muse.cli.config import LimitsConfig, MuseConfig, _dump_toml
227	cfg: MuseConfig = {"limits": LimitsConfig(shard_prefix_length=4)}
228	toml_text = _dump_toml(cfg)
229	parsed = tomllib.loads(toml_text)
230	assert parsed["limits"]["shard_prefix_length"] == 4
231
232	def test_domain_key_injection_blocked_in_dump(self) -> None:
233	from muse.cli.config import MuseConfig, _dump_toml
234	cfg: MuseConfig = {"domain": {"malicious\nkey": "val"}}
235	with pytest.raises(ValueError, match="TOML keys"):
236	_dump_toml(cfg)
237
238	def test_remote_name_injection_blocked(self) -> None:
239	from muse.cli.config import MuseConfig, RemoteEntry, _dump_toml
240	cfg: MuseConfig = {"remotes": {"malicious\nname": RemoteEntry(url="http://localhost")}}
241	with pytest.raises(ValueError, match="TOML keys"):
242	_dump_toml(cfg)
243
244	def test_value_with_newline_escaped_not_injected(self) -> None:
245	from muse.cli.config import MuseConfig, _dump_toml
246	cfg: MuseConfig = {"domain": {"key": "line1\nline2"}}
247	toml_text = _dump_toml(cfg)
248	parsed = tomllib.loads(toml_text)
249	# The value line should contain \\n (escaped), not a literal newline
250	value_line = next(l for l in toml_text.splitlines() if l.startswith("key"))
251	assert "\n" not in value_line
252	assert "\\n" in value_line
253	assert "line1" in parsed["domain"]["key"]
254
255	def test_user_section_not_emitted(self) -> None:
256	"""_dump_toml must never emit a [user] section — identity lives in identity.toml."""
257	from muse.cli.config import HubConfig, MuseConfig, _dump_toml
258	cfg: MuseConfig = {
259	"hub": HubConfig(url="https://musehub.ai"),
260	"domain": {"k": "v"},
261	}
262	toml_text = _dump_toml(cfg)
263	assert "[user]" not in toml_text
264
265	def test_hub_before_domain_in_section_order(self) -> None:
266	from muse.cli.config import HubConfig, MuseConfig, _dump_toml
267	cfg: MuseConfig = {
268	"hub": HubConfig(url="https://musehub.ai"),
269	"domain": {"k": "v"},
270	}
271	toml_text = _dump_toml(cfg)
272	hub_pos = toml_text.index("[hub]")
273	domain_pos = toml_text.index("[domain]")
274	assert hub_pos < domain_pos
275
276
277	# ── Unit: set_config_value + get_config_value ─────────────────────────────────
278
279
280	class TestSetGetConfigValue:
281	def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
282	dot_muse = muse_dir(tmp_path)
283	dot_muse.mkdir()
284	(dot_muse / "config.toml").write_text("")
285	return tmp_path
286
287	def test_limits_max_walk_commits_writes_to_limits_section(
288	self, tmp_path: pathlib.Path
289	) -> None:
290	root = self._make_repo(tmp_path)
291	from muse.cli.config import set_config_value
292	set_config_value("limits.max_walk_commits", "5000", root)
293	raw = (config_toml_path(root)).read_text()
294	parsed = tomllib.loads(raw)
295	assert "limits" in parsed
296	assert parsed["limits"]["max_walk_commits"] == 5000
297	assert "domain" not in parsed
298
299	def test_limits_max_walk_commits_NOT_written_to_domain(
300	self, tmp_path: pathlib.Path
301	) -> None:
302	"""Regression: previously limits fell through to domain code path."""
303	root = self._make_repo(tmp_path)
304	from muse.cli.config import set_config_value
305	set_config_value("limits.max_walk_commits", "1000", root)
306	raw = (config_toml_path(root)).read_text()
307	parsed = tomllib.loads(raw)
308	assert "domain" not in parsed
309
310	def test_limits_shard_prefix_length_valid(self, tmp_path: pathlib.Path) -> None:
311	root = self._make_repo(tmp_path)
312	from muse.cli.config import get_config_value, set_config_value
313	set_config_value("limits.shard_prefix_length", "4", root)
314	assert get_config_value("limits.shard_prefix_length", root) == "4"
315
316	def test_limits_shard_prefix_length_invalid_rejected(
317	self, tmp_path: pathlib.Path
318	) -> None:
319	root = self._make_repo(tmp_path)
320	from muse.cli.config import set_config_value
321	with pytest.raises(ValueError, match="shard_prefix_length must be 2 or 4"):
322	set_config_value("limits.shard_prefix_length", "3", root)
323
324	def test_limits_non_integer_rejected(self, tmp_path: pathlib.Path) -> None:
325	root = self._make_repo(tmp_path)
326	from muse.cli.config import set_config_value
327	with pytest.raises(ValueError, match="integer"):
328	set_config_value("limits.max_walk_commits", "notanint", root)
329
330	def test_limits_zero_rejected(self, tmp_path: pathlib.Path) -> None:
331	root = self._make_repo(tmp_path)
332	from muse.cli.config import set_config_value
333	with pytest.raises(ValueError, match="positive"):
334	set_config_value("limits.max_walk_commits", "0", root)
335
336	def test_limits_negative_rejected(self, tmp_path: pathlib.Path) -> None:
337	root = self._make_repo(tmp_path)
338	from muse.cli.config import set_config_value
339	with pytest.raises(ValueError, match="positive"):
340	set_config_value("limits.max_walk_commits", "-1", root)
341
342	def test_limits_unknown_key_rejected(self, tmp_path: pathlib.Path) -> None:
343	root = self._make_repo(tmp_path)
344	from muse.cli.config import set_config_value
345	with pytest.raises(ValueError, match="Unknown \\[limits\\]"):
346	set_config_value("limits.unknown_key", "5", root)
347
348	def test_domain_key_injection_rejected(self, tmp_path: pathlib.Path) -> None:
349	root = self._make_repo(tmp_path)
350	from muse.cli.config import set_config_value
351	with pytest.raises(ValueError, match="TOML keys"):
352	set_config_value("domain.malicious\nkey", "bad", root)
353
354	def test_domain_key_with_bracket_injection_rejected(
355	self, tmp_path: pathlib.Path
356	) -> None:
357	root = self._make_repo(tmp_path)
358	from muse.cli.config import set_config_value
359	with pytest.raises(ValueError, match="TOML keys"):
360	set_config_value("domain.x][malicious", "bad", root)
361
362	def test_domain_safe_key_written(self, tmp_path: pathlib.Path) -> None:
363	root = self._make_repo(tmp_path)
364	from muse.cli.config import get_config_value, set_config_value
365	set_config_value("domain.ticks_per_beat", "480", root)
366	assert get_config_value("domain.ticks_per_beat", root) == "480"
367
368	def test_get_config_value_limits_after_write(self, tmp_path: pathlib.Path) -> None:
369	root = self._make_repo(tmp_path)
370	from muse.cli.config import get_config_value, set_config_value
371	set_config_value("limits.max_ancestors", "25000", root)
372	assert get_config_value("limits.max_ancestors", root) == "25000"
373
374	def test_user_namespace_raises_value_error(self, tmp_path: pathlib.Path) -> None:
375	"""user.* writes must be blocked — identity lives in identity.toml."""
376	root = self._make_repo(tmp_path)
377	from muse.cli.config import set_config_value
378	with pytest.raises(ValueError):
379	set_config_value("user.handle", "alice", root)
380
381	def test_user_email_raises_value_error(self, tmp_path: pathlib.Path) -> None:
382	root = self._make_repo(tmp_path)
383	from muse.cli.config import set_config_value
384	with pytest.raises(ValueError):
385	set_config_value("user.email", "[email protected]", root)
386
387	def test_user_type_raises_value_error(self, tmp_path: pathlib.Path) -> None:
388	root = self._make_repo(tmp_path)
389	from muse.cli.config import set_config_value
390	with pytest.raises(ValueError):
391	set_config_value("user.type", "human", root)
392
393
394	# ── Unit: config_as_dict ─────────────────────────────────────────────────────
395
396
397	class TestConfigAsDict:
398	def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
399	dot_muse = muse_dir(tmp_path)
400	dot_muse.mkdir()
401	return tmp_path
402
403	def test_limits_included_in_output(self, tmp_path: pathlib.Path) -> None:
404	root = self._make_repo(tmp_path)
405	(config_toml_path(root)).write_text(
406	"[limits]\nmax_walk_commits = 5000\n"
407	)
408	from muse.cli.config import config_as_dict
409	d = config_as_dict(root)
410	assert "limits" in d
411	assert d["limits"]["max_walk_commits"] == "5000"
412
413	def test_limits_absent_when_not_set(self, tmp_path: pathlib.Path) -> None:
414	root = self._make_repo(tmp_path)
415	# Use a [hub] section to populate the config without [limits]
416	(config_toml_path(root)).write_text('[hub]\nurl = "https://musehub.ai"\n')
417	from muse.cli.config import config_as_dict
418	d = config_as_dict(root)
419	assert "limits" not in d
420
421	def test_empty_config_returns_empty_dict(self, tmp_path: pathlib.Path) -> None:
422	root = self._make_repo(tmp_path)
423	(config_toml_path(root)).write_text("")
424	from muse.cli.config import config_as_dict
425	assert config_as_dict(root) == {}
426
427	def test_user_section_silently_dropped(self, tmp_path: pathlib.Path) -> None:
428	"""Old config.toml files with a [user] section must have it silently dropped."""
429	root = self._make_repo(tmp_path)
430	(config_toml_path(root)).write_text(
431	'[user]\nhandle = "alice"\n'
432	)
433	from muse.cli.config import config_as_dict
434	d = config_as_dict(root)
435	assert "user" not in d
436
437
438	# ── Integration: run_show ─────────────────────────────────────────────────────
439
440
441	class TestRunRead:
442	def test_read_json_includes_limits(self, repo: pathlib.Path) -> None:
443	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "7777"])
444	result = runner.invoke(cli, ["config", "read", "--json"])
445	assert result.exit_code == 0
446	data = _json_show(result)
447	assert "limits" in data
448	limits = data["limits"]
449	assert isinstance(limits, dict)
450	assert limits["max_walk_commits"] == "7777"
451
452	def test_read_json_schema_no_user_section(self, repo: pathlib.Path) -> None:
453	"""[user] section must be absent from config read --json output.
454
455	User identity now lives exclusively in identity.toml; config.toml
456	no longer stores or emits a [user] section.
457	"""
458	result = runner.invoke(cli, ["config", "read", "--json"])
459	assert result.exit_code == 0
460	data = _json_show(result)
461	assert "user" not in data
462
463	def test_read_json_flag_emits_json(self, repo: pathlib.Path) -> None:
464	result = runner.invoke(cli, ["config", "read", "--json"])
465	assert result.exit_code == 0
466	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
467	assert len(json_lines) >= 1
468
469	def test_read_format_invalid_exits(self, repo: pathlib.Path) -> None:
470	result = runner.invoke(cli, ["config", "read", "--format", "xml"])
471	assert result.exit_code != 0
472
473	def test_read_text_mode_ansi_sanitized(
474	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
475	) -> None:
476	# Write a config with ANSI in value (directly to file to bypass validation)
477	(config_toml_path(repo)).write_text(
478	'[domain]\nticks = "\\x1b[31mmalicious\\x1b[0m"\n'
479	)
480	result = runner.invoke(cli, ["config", "read"])
481	assert "\x1b[" not in result.output
482
483	def test_read_text_empty_config(self, repo: pathlib.Path) -> None:
484	result = runner.invoke(cli, ["config", "read"])
485	assert result.exit_code == 0
486	assert "No configuration set" in result.output
487
488	def test_read_text_limits_section_displayed(self, repo: pathlib.Path) -> None:
489	runner.invoke(cli, ["config", "set", "limits.max_ancestors", "30000"])
490	result = runner.invoke(cli, ["config", "read"])
491	assert result.exit_code == 0
492	assert "[limits]" in result.output
493	assert "max_ancestors" in result.output
494
495	def test_read_json_stdout_clean(self, repo: pathlib.Path) -> None:
496	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
497	result = runner.invoke(cli, ["config", "read", "--json"])
498	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
499	assert len(json_lines) >= 1
500
501	def test_read_no_repo_still_works(
502	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
503	) -> None:
504	# show gracefully shows empty config outside a repo (uses cwd)
505	monkeypatch.chdir(tmp_path)
506	result = runner.invoke(cli, ["config", "read"])
507	assert result.exit_code == 0
508
509
510	# ── Integration: run_get ─────────────────────────────────────────────────────
511
512
513	class TestRunGet:
514	def test_get_existing_hub_key_raw_value(self, repo: pathlib.Path) -> None:
515	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
516	result = runner.invoke(cli, ["config", "get", "hub.url"])
517	assert result.exit_code == 0
518	assert "musehub.ai" in result.output
519
520	def test_get_json_schema_domain_key(self, repo: pathlib.Path) -> None:
521	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
522	result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat", "--json"])
523	assert result.exit_code == 0
524	data = _json_get(result)
525	assert data["key"] == "domain.ticks_per_beat"
526	assert data["value"] == "480"
527
528	def test_get_missing_key_exits_nonzero(self, repo: pathlib.Path) -> None:
529	result = runner.invoke(cli, ["config", "get", "hub.url"])
530	assert result.exit_code != 0
531
532	def test_get_missing_key_error_to_stderr(
533	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
534	) -> None:
535	result = runner.invoke(cli, ["config", "get", "hub.url"])
536	assert result.exit_code != 0
537	assert "not set" in result.stderr
538
539	def test_get_limits_key_after_set(self, repo: pathlib.Path) -> None:
540	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "12345"])
541	result = runner.invoke(cli, ["config", "get", "limits.max_walk_commits"])
542	assert result.exit_code == 0
543	assert "12345" in result.output
544
545	def test_get_json_stdout_clean(self, repo: pathlib.Path) -> None:
546	runner.invoke(cli, ["config", "set", "domain.sample_rate", "44100"])
547	result = runner.invoke(cli, ["config", "get", "domain.sample_rate", "--json"])
548	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
549	assert len(json_lines) >= 1
550
551	def test_get_user_namespace_exits_nonzero(self, repo: pathlib.Path) -> None:
552	"""user.* get must exit nonzero — user identity is not stored in config.toml."""
553	result = runner.invoke(cli, ["config", "get", "user.handle"])
554	assert result.exit_code != 0
555
556
557	# ── Integration: run_set ─────────────────────────────────────────────────────
558
559
560	class TestRunSet:
561	def test_set_success_json_schema_domain(self, repo: pathlib.Path) -> None:
562	result = runner.invoke(
563	cli, ["config", "set", "domain.ticks_per_beat", "960", "--json"]
564	)
565	assert result.exit_code == 0
566	data = _json_set(result)
567	assert data["status"] == "ok"
568	assert data["key"] == "domain.ticks_per_beat"
569	assert data["value"] == "960"
570
571	def test_set_success_stderr_message_domain(
572	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
573	) -> None:
574	result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
575	assert result.exit_code == 0
576	assert "960" in result.stderr
577
578	def test_set_json_stdout_clean(self, repo: pathlib.Path) -> None:
579	result = runner.invoke(
580	cli, ["config", "set", "domain.ticks_per_beat", "480", "--json"]
581	)
582	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
583	assert len(json_lines) >= 1
584
585	def test_set_user_namespace_blocked(self, repo: pathlib.Path) -> None:
586	"""user.* writes are blocked — identity lives in identity.toml."""
587	result = runner.invoke(cli, ["config", "set", "user.handle", "Alice"])
588	assert result.exit_code != 0
589
590	def test_set_user_namespace_error_mentions_auth(self, repo: pathlib.Path) -> None:
591	"""Error message for user.* must guide user toward auth/identity commands."""
592	result = runner.invoke(cli, ["config", "set", "user.handle", "Alice"])
593	assert result.exit_code != 0
594	assert "auth" in result.stderr.lower() or "identity" in result.stderr.lower()
595
596	def test_set_blocked_namespace_exits(self, repo: pathlib.Path) -> None:
597	result = runner.invoke(cli, ["config", "set", "auth.token", "secret"])
598	assert result.exit_code != 0
599
600	def test_set_blocked_remotes_namespace_exits(self, repo: pathlib.Path) -> None:
601	result = runner.invoke(cli, ["config", "set", "remotes.origin", "url"])
602	assert result.exit_code != 0
603
604	def test_set_domain_newline_injection_rejected(self, repo: pathlib.Path) -> None:
605	result = runner.invoke(cli, ["config", "set", "domain.malicious\nkey", "bad"])
606	assert result.exit_code != 0
607
608	def test_set_domain_bracket_injection_rejected(self, repo: pathlib.Path) -> None:
609	result = runner.invoke(cli, ["config", "set", "domain.x][malicious", "bad"])
610	assert result.exit_code != 0
611
612	def test_set_limits_max_walk_commits(self, repo: pathlib.Path) -> None:
613	result = runner.invoke(
614	cli, ["config", "set", "limits.max_walk_commits", "20000"]
615	)
616	assert result.exit_code == 0
617	get_result = runner.invoke(cli, ["config", "get", "limits.max_walk_commits"])
618	assert "20000" in get_result.output
619
620	def test_set_limits_non_integer_rejected(self, repo: pathlib.Path) -> None:
621	result = runner.invoke(
622	cli, ["config", "set", "limits.max_walk_commits", "abc"]
623	)
624	assert result.exit_code != 0
625
626	def test_set_limits_zero_rejected(self, repo: pathlib.Path) -> None:
627	result = runner.invoke(
628	cli, ["config", "set", "limits.max_walk_commits", "0"]
629	)
630	assert result.exit_code != 0
631
632	def test_set_limits_shard_prefix_valid(self, repo: pathlib.Path) -> None:
633	result = runner.invoke(
634	cli, ["config", "set", "limits.shard_prefix_length", "4"]
635	)
636	assert result.exit_code == 0
637
638	def test_set_limits_shard_prefix_invalid_rejected(self, repo: pathlib.Path) -> None:
639	result = runner.invoke(
640	cli, ["config", "set", "limits.shard_prefix_length", "3"]
641	)
642	assert result.exit_code != 0
643
644	def test_set_error_ansi_sanitized_in_output(
645	self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
646	) -> None:
647	# Use a key with ANSI that would appear in the error message
648	ansi_key = "domain.\x1b[31mmalicious\x1b[0m\nkey"
649	result = runner.invoke(cli, ["config", "set", ansi_key, "val"])
650	assert result.exit_code != 0
651	assert "\x1b[" not in result.output
652
653	def test_set_limits_writes_to_limits_not_domain(self, repo: pathlib.Path) -> None:
654	"""Regression: limits namespace must not fall through to domain code."""
655	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "9999"])
656	raw = (config_toml_path(repo)).read_text()
657	parsed = tomllib.loads(raw)
658	assert "domain" not in parsed
659	assert parsed["limits"]["max_walk_commits"] == 9999
660
661
662	# ── Integration: run_edit ─────────────────────────────────────────────────────
663
664
665	class TestRunEdit:
666	def test_edit_no_repo_exits(
667	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
668	) -> None:
669	monkeypatch.chdir(tmp_path)
670	result = runner.invoke(cli, ["config", "edit"])
671	assert result.exit_code != 0
672
673	def test_edit_missing_config_file_autocreated(
674	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
675	) -> None:
676	"""Missing config.toml must be auto-created before the editor opens."""
677	(config_toml_path(repo)).unlink()
678	monkeypatch.setenv("EDITOR", "true")
679	monkeypatch.delenv("VISUAL", raising=False)
680	result = runner.invoke(cli, ["config", "edit"])
681	assert result.exit_code == 0
682	assert (config_toml_path(repo)).exists()
683
684	def test_edit_bad_editor_exits(
685	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
686	) -> None:
687	monkeypatch.setenv("EDITOR", "nonexistent-editor-xyz")
688	monkeypatch.delenv("VISUAL", raising=False)
689	result = runner.invoke(cli, ["config", "edit"])
690	assert result.exit_code != 0
691
692	def test_edit_invokes_editor(
693	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
694	) -> None:
695	monkeypatch.setenv("EDITOR", "true")
696	monkeypatch.delenv("VISUAL", raising=False)
697	result = runner.invoke(cli, ["config", "edit"])
698	assert result.exit_code == 0
699
700
701	# ── Security ──────────────────────────────────────────────────────────────────
702
703
704	class TestConfigSecurity:
705	def test_toml_key_injection_blocked_end_to_end(self, repo: pathlib.Path) -> None:
706	"""Setting domain key with newline must not corrupt config.toml."""
707	result = runner.invoke(
708	cli, ["config", "set", "domain.malicious\nkey", "bad"]
709	)
710	assert result.exit_code != 0
711	raw = (config_toml_path(repo)).read_text()
712	assert "\nkey" not in raw
713
714	def test_bracket_injection_blocked(self, repo: pathlib.Path) -> None:
715	result = runner.invoke(
716	cli, ["config", "set", "domain.x][malicious", "val"]
717	)
718	assert result.exit_code != 0
719	raw = (config_toml_path(repo)).read_text()
720	assert "[malicious]" not in raw
721
722	def test_equals_injection_blocked(self, repo: pathlib.Path) -> None:
723	result = runner.invoke(
724	cli, ["config", "set", "domain.x=y", "val"]
725	)
726	assert result.exit_code != 0
727
728	def test_ansi_in_read_text_stripped(self, repo: pathlib.Path) -> None:
729	(config_toml_path(repo)).write_text(
730	'[domain]\nticks = "\\x1b[31mred\\x1b[0m"\n'
731	)
732	result = runner.invoke(cli, ["config", "read"])
733	assert "\x1b[" not in result.output
734
735	def test_auth_namespace_always_blocked(self, repo: pathlib.Path) -> None:
736	for key in ("auth.token", "auth.password", "auth.secret"):
737	result = runner.invoke(cli, ["config", "set", key, "val"])
738	assert result.exit_code != 0, f"Expected {key!r} to be blocked"
739
740	def test_user_namespace_always_blocked(self, repo: pathlib.Path) -> None:
741	"""user.* writes must always be blocked; identity.toml owns that namespace."""
742	for key in ("user.handle", "user.email", "user.type", "user.display_name"):
743	result = runner.invoke(cli, ["config", "set", key, "val"])
744	assert result.exit_code != 0, f"Expected {key!r} to be blocked"
745
746	def test_credentials_not_in_json_output(self, repo: pathlib.Path) -> None:
747	(config_toml_path(repo)).write_text(
748	'[hub]\nurl = "https://localhost:1337"\n'
749	'[auth]\ntoken = "secret-token"\n'
750	)
751	result = runner.invoke(cli, ["config", "read", "--json"])
752	assert result.exit_code == 0
753	assert "secret-token" not in result.output
754
755	def test_value_newline_escaped_in_toml(self, repo: pathlib.Path) -> None:
756	runner.invoke(cli, ["config", "set", "domain.label", "line1\nline2"])
757	raw = (config_toml_path(repo)).read_text()
758	# The literal newline must not appear in the value field
759	label_line = [l for l in raw.splitlines() if "label" in l][0]
760	assert "\n" not in label_line
761
762	def test_error_message_to_stderr_not_stdout(self, repo: pathlib.Path) -> None:
763	result = runner.invoke(
764	cli, ["config", "set", "domain.malicious\nkey", "val"]
765	)
766	assert result.exit_code != 0
767	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
768	assert len(json_lines) == 0
769
770	def test_user_section_dropped_from_old_config(self, repo: pathlib.Path) -> None:
771	"""Old config.toml with [user] section must not expose it in read output."""
772	(config_toml_path(repo)).write_text(
773	'[user]\nhandle = "alice"\n'
774	)
775	result = runner.invoke(cli, ["config", "read", "--json"])
776	assert result.exit_code == 0
777	data = _json_show(result)
778	assert "user" not in data
779
780
781	# ── E2E round-trips ───────────────────────────────────────────────────────────
782
783
784	class TestE2ERoundTrips:
785	def test_set_user_namespace_blocked_end_to_end(self, repo: pathlib.Path) -> None:
786	"""user.* is blocked at the CLI layer — exit nonzero, nothing written."""
787	result = runner.invoke(cli, ["config", "set", "user.handle", "DeepBlue"])
788	assert result.exit_code != 0
789	raw = (config_toml_path(repo)).read_text()
790	assert "DeepBlue" not in raw
791
792	def test_set_hub_then_get(self, repo: pathlib.Path) -> None:
793	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
794	result = runner.invoke(cli, ["config", "get", "hub.url"])
795	assert result.exit_code == 0
796	assert "musehub.ai" in result.output
797
798	def test_set_limits_then_read_json(self, repo: pathlib.Path) -> None:
799	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "3333"])
800	result = runner.invoke(cli, ["config", "read", "--json"])
801	assert result.exit_code == 0
802	data = _json_show(result)
803	limits = data.get("limits")
804	assert isinstance(limits, dict)
805	assert limits.get("max_walk_commits") == "3333"
806
807	def test_set_domain_then_read_valid_toml(self, repo: pathlib.Path) -> None:
808	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
809	result = runner.invoke(cli, ["config", "read"])
810	assert result.exit_code == 0
811	assert "960" in result.output
812
813	def test_limits_written_to_limits_section_not_domain(
814	self, repo: pathlib.Path
815	) -> None:
816	runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "8888"])
817	raw = (config_toml_path(repo)).read_text()
818	parsed = tomllib.loads(raw)
819	assert "domain" not in parsed
820	assert parsed["limits"]["max_graph_commits"] == 8888
821
822	def test_multiple_writes_preserve_all_sections(self, repo: pathlib.Path) -> None:
823	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
824	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
825	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "2000"])
826	raw = (config_toml_path(repo)).read_text()
827	parsed = tomllib.loads(raw)
828	assert parsed["hub"]["url"] == "https://musehub.ai"
829	assert parsed["domain"]["ticks_per_beat"] == "480"
830	assert parsed["limits"]["max_walk_commits"] == 2000
831
832	def test_set_json_get_json_consistent_domain(self, repo: pathlib.Path) -> None:
833	set_result = runner.invoke(
834	cli, ["config", "set", "domain.ticks_per_beat", "960", "--json"]
835	)
836	get_result = runner.invoke(
837	cli, ["config", "get", "domain.ticks_per_beat", "--json"]
838	)
839	set_data = _json_set(set_result)
840	get_data = _json_get(get_result)
841	assert set_data["value"] == get_data["value"] == "960"
842
843
844	# ── Stress ────────────────────────────────────────────────────────────────────
845
846
847	class TestStress:
848	def test_8_concurrent_set_domain_to_isolated_repos(
849	self, tmp_path: pathlib.Path
850	) -> None:
851	"""8 threads writing domain keys to independent repos must not corrupt each other."""
852	from muse._version import __version__
853	from muse.cli.config import get_config_value, set_config_value
854
855	errors: list[str] = []
856
857	def _do(idx: int) -> None:
858	try:
859	repo_dir = tmp_path / f"repo_{idx}"
860	dot_muse = muse_dir(repo_dir)
861	dot_muse.mkdir(parents=True)
862	(dot_muse / "config.toml").write_text("")
863	(dot_muse / "repo.json").write_text(
864	json.dumps({
865	"repo_id": f"repo-{idx}",
866	"schema_version": __version__,
867	"domain": "code",
868	})
869	)
870	value = f"val_{idx}"
871	set_config_value("domain.label", value, repo_dir)
872	result = get_config_value("domain.label", repo_dir)
873	assert result == value, f"Expected {value!r}, got {result!r}"
874	except Exception as exc:
875	errors.append(f"Thread {idx}: {exc}")
876
877	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
878	for t in threads:
879	t.start()
880	for t in threads:
881	t.join()
882	assert errors == [], f"Concurrent config write failures:\n{'\n'.join(errors)}"
883
884	def test_8_concurrent_user_set_all_blocked(
885	self, tmp_path: pathlib.Path
886	) -> None:
887	"""8 threads attempting user.* writes must all raise ValueError."""
888	from muse._version import __version__
889	from muse.cli.config import set_config_value
890
891	errors: list[str] = []
892
893	def _do(idx: int) -> None:
894	try:
895	repo_dir = tmp_path / f"user_repo_{idx}"
896	dot_muse = muse_dir(repo_dir)
897	dot_muse.mkdir(parents=True)
898	(dot_muse / "config.toml").write_text("")
899	(dot_muse / "repo.json").write_text(
900	json.dumps({
901	"repo_id": f"repo-{idx}",
902	"schema_version": __version__,
903	"domain": "code",
904	})
905	)
906	try:
907	set_config_value("user.handle", f"user_{idx}", repo_dir)
908	errors.append(f"Thread {idx}: expected ValueError, got none")
909	except ValueError:
910	pass # expected
911	except Exception as exc:
912	errors.append(f"Thread {idx}: unexpected error: {exc}")
913
914	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
915	for t in threads:
916	t.start()
917	for t in threads:
918	t.join()
919	assert errors == [], f"Unexpected results in user.* block tests:\n{'\n'.join(errors)}"
920
921	def test_8_concurrent_set_limits_isolated(
922	self, tmp_path: pathlib.Path
923	) -> None:
924	"""8 threads writing limits to isolated repos must write to [limits] not [domain]."""
925	from muse._version import __version__
926	from muse.cli.config import set_config_value
927
928	errors: list[str] = []
929
930	def _do(idx: int) -> None:
931	try:
932	repo_dir = tmp_path / f"limits_repo_{idx}"
933	dot_muse = muse_dir(repo_dir)
934	dot_muse.mkdir(parents=True)
935	(dot_muse / "config.toml").write_text("")
936	(dot_muse / "repo.json").write_text(
937	json.dumps({
938	"repo_id": f"repo-{idx}",
939	"schema_version": __version__,
940	"domain": "code",
941	})
942	)
943	set_config_value("limits.max_walk_commits", str(1000 + idx), repo_dir)
944	raw = (dot_muse / "config.toml").read_text()
945	parsed = tomllib.loads(raw)
946	assert "limits" in parsed, "limits section missing"
947	assert "domain" not in parsed, "limits fell through to domain"
948	assert parsed["limits"]["max_walk_commits"] == 1000 + idx
949	except Exception as exc:
950	errors.append(f"Thread {idx}: {exc}")
951
952	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
953	for t in threads:
954	t.start()
955	for t in threads:
956	t.join()
957	assert errors == [], f"Concurrent limits write failures:\n{'\n'.join(errors)}"
958
959
960	# =============================================================================
961	# muse config read — extended hardening
962	# =============================================================================
963
964
965	class TestRunReadExtended:
966	"""Additional coverage for ``muse config read`` gaps."""
967
968	# ── flag aliases ──────────────────────────────────────────────────────────
969
970	def test_j_short_flag_emits_json(self, repo: pathlib.Path) -> None:
971	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
972	result = runner.invoke(cli, ["config", "read", "-j"])
973	assert result.exit_code == 0
974	data = _json_show(result)
975	assert isinstance(data, dict)
976
977	def test_j_flag_same_as_json_flag(self, repo: pathlib.Path) -> None:
978	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
979	r1 = runner.invoke(cli, ["config", "read", "--json"])
980	r2 = runner.invoke(cli, ["config", "read", "-j"])
981	assert r1.exit_code == 0
982	assert r2.exit_code == 0
983	# Compare only config sections, not timing fields which naturally differ
984	d1 = {k: v for k, v in _json_show(r1).items() if k not in ("duration_ms", "exit_code")}
985	d2 = {k: v for k, v in _json_show(r2).items() if k not in ("duration_ms", "exit_code")}
986	assert d1 == d2
987
988	def test_j_shorthand_emits_json(self, repo: pathlib.Path) -> None:
989	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
990	result = runner.invoke(cli, ["config", "read", "-j"])
991	assert result.exit_code == 0
992	data = _json_show(result)
993	assert isinstance(data, dict)
994
995	def test_default_is_text_output(self, repo: pathlib.Path) -> None:
996	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
997	result = runner.invoke(cli, ["config", "read"])
998	assert result.exit_code == 0
999	assert "[hub]" in result.output
1000	assert "{" not in result.output # no JSON
1001
1002	def test_json_flag_emits_single_object(
1003	self, repo: pathlib.Path
1004	) -> None:
1005	"""--json must emit exactly one JSON object, not duplicate output."""
1006	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1007	result = runner.invoke(cli, ["config", "read", "--json"])
1008	assert result.exit_code == 0
1009	# Must be parseable as a single JSON object
1010	data = _json_show(result)
1011	assert isinstance(data, dict)
1012
1013	# ── JSON structure ────────────────────────────────────────────────────────
1014
1015	def test_json_is_compact(self, repo: pathlib.Path) -> None:
1016	"""JSON output must be compact — agents parse it, not humans."""
1017	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1018	result = runner.invoke(cli, ["config", "read", "--json"])
1019	assert result.exit_code == 0
1020	# Compact JSON is a single line
1021	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1022	assert len(json_lines) == 1
1023	json.loads(json_lines[0]) # must be valid JSON
1024
1025	def test_json_hub_section_present(self, repo: pathlib.Path) -> None:
1026	from muse.cli.config import set_hub_url
1027	set_hub_url("https://musehub.ai", repo)
1028	result = runner.invoke(cli, ["config", "read", "--json"])
1029	assert result.exit_code == 0
1030	data = _json_show(result)
1031	assert "hub" in data
1032	hub = data["hub"]
1033	assert isinstance(hub, dict)
1034	assert hub["url"] == "https://musehub.ai"
1035
1036	def test_json_remotes_section_present(self, repo: pathlib.Path) -> None:
1037	from muse.cli.config import set_remote
1038	set_remote("origin", "https://hub.example.com/owner/repo", repo)
1039	result = runner.invoke(cli, ["config", "read", "--json"])
1040	assert result.exit_code == 0
1041	data = _json_show(result)
1042	assert "remotes" in data
1043	remotes = data["remotes"]
1044	assert isinstance(remotes, dict)
1045	assert "origin" in remotes
1046
1047	def test_json_domain_section_present(self, repo: pathlib.Path) -> None:
1048	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
1049	result = runner.invoke(cli, ["config", "read", "--json"])
1050	assert result.exit_code == 0
1051	data = _json_show(result)
1052	assert "domain" in data
1053	domain = data["domain"]
1054	assert isinstance(domain, dict)
1055	assert domain["ticks_per_beat"] == "960"
1056
1057	def test_json_no_user_section(self, repo: pathlib.Path) -> None:
1058	"""[user] must never appear in config read --json output."""
1059	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1060	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "5000"])
1061	result = runner.invoke(cli, ["config", "read", "--json"])
1062	assert result.exit_code == 0
1063	data = _json_show(result)
1064	assert "user" not in data
1065	assert "domain" in data
1066	assert "limits" in data
1067
1068	def test_json_empty_config_has_no_sections(self, repo: pathlib.Path) -> None:
1069	result = runner.invoke(cli, ["config", "read", "--json"])
1070	assert result.exit_code == 0
1071	data = _json_show(result)
1072	# Config sections absent; only timing/status keys present
1073	for section in ("user", "hub", "remotes", "domain", "limits"):
1074	assert section not in data
1075
1076	def test_json_no_credentials(self, repo: pathlib.Path) -> None:
1077	"""Credentials (auth, token) must never appear in JSON output."""
1078	# Write a config with a fake [auth] section directly
1079	(config_toml_path(repo)).write_text(
1080	'[hub]\nurl = "https://musehub.ai"\n\n[auth]\ntoken = "secret"\n'
1081	)
1082	result = runner.invoke(cli, ["config", "read", "--json"])
1083	assert result.exit_code == 0
1084	assert "secret" not in result.output
1085	assert "token" not in result.output
1086
1087	# ── text mode structure ───────────────────────────────────────────────────
1088
1089	def test_text_output_to_stdout(self, repo: pathlib.Path) -> None:
1090	"""Text mode config content goes to stdout, not only stderr."""
1091	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1092	result = runner.invoke(cli, ["config", "read"])
1093	assert result.exit_code == 0
1094	assert "[hub]" in result.output
1095
1096	def test_text_no_user_section_header(self, repo: pathlib.Path) -> None:
1097	"""Text mode must never emit [user] — user identity is not in config.toml."""
1098	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1099	result = runner.invoke(cli, ["config", "read"])
1100	assert "[user]" not in result.output
1101
1102	def test_text_hub_section_header(self, repo: pathlib.Path) -> None:
1103	from muse.cli.config import set_hub_url
1104	set_hub_url("https://musehub.ai", repo)
1105	result = runner.invoke(cli, ["config", "read"])
1106	assert "[hub]" in result.output
1107
1108	def test_text_remotes_section_header(self, repo: pathlib.Path) -> None:
1109	from muse.cli.config import set_remote
1110	set_remote("origin", "https://hub.example.com/owner/repo", repo)
1111	result = runner.invoke(cli, ["config", "read"])
1112	assert "[remotes." in result.output
1113
1114	def test_text_domain_section_header(self, repo: pathlib.Path) -> None:
1115	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1116	result = runner.invoke(cli, ["config", "read"])
1117	assert "[domain]" in result.output
1118
1119	def test_text_key_value_format_domain(self, repo: pathlib.Path) -> None:
1120	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1121	result = runner.invoke(cli, ["config", "read"])
1122	# TOML key = "value" format
1123	assert 'ticks_per_beat = "480"' in result.output
1124
1125	def test_text_limits_no_quotes_on_values(self, repo: pathlib.Path) -> None:
1126	"""Limits values are integers in TOML — no quotes."""
1127	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "8000"])
1128	result = runner.invoke(cli, ["config", "read"])
1129	assert "max_walk_commits = 8000" in result.output
1130
1131	def test_read_help_contains_quickstart(self) -> None:
1132	result = runner.invoke(cli, ["config", "read", "--help"])
1133	assert result.exit_code == 0
1134	assert "quickstart" in result.output.lower() or "jq" in result.output
1135
1136	def test_read_help_contains_exit_codes(self) -> None:
1137	result = runner.invoke(cli, ["config", "read", "--help"])
1138	assert result.exit_code == 0
1139	assert "Exit codes" in result.output or "exit" in result.output.lower()
1140
1141
1142	class TestRunReadStress:
1143	"""Stress tests for ``muse config read``."""
1144
1145	def test_concurrent_read_calls(self, repo: pathlib.Path) -> None:
1146	"""Concurrent config_as_dict reads of the same config must all succeed.
1147
1148	Uses config_as_dict directly — CliRunner shares a buffer across threads
1149	so full CLI invocations cannot be called concurrently from different threads.
1150	"""
1151	import threading
1152	from muse.cli.config import config_as_dict, set_config_value
1153	set_config_value("hub.url", "https://musehub.ai", repo)
1154	set_config_value("limits.max_walk_commits", "1000", repo)
1155	errors: list[str] = []
1156
1157	def _do(idx: int) -> None:
1158	try:
1159	data = config_as_dict(repo)
1160	assert "hub" in data
1161	assert data["hub"]["url"] == "https://musehub.ai"
1162	except Exception as exc:
1163	errors.append(f"Thread {idx}: {exc}")
1164
1165	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1166	for t in threads:
1167	t.start()
1168	for t in threads:
1169	t.join()
1170	assert errors == [], "\n".join(errors)
1171
1172	def test_read_large_domain_config(self, repo: pathlib.Path) -> None:
1173	"""Read handles a config with many domain keys without truncation."""
1174	for i in range(20):
1175	runner.invoke(cli, ["config", "set", f"domain.key_{i}", str(i * 10)])
1176	result = runner.invoke(cli, ["config", "read", "--json"])
1177	assert result.exit_code == 0
1178	data = _json_show(result)
1179	domain = data.get("domain")
1180	assert isinstance(domain, dict)
1181	assert len(domain) == 20
1182
1183	def test_read_json_output_is_valid_json(self, repo: pathlib.Path) -> None:
1184	"""JSON output must always be parseable regardless of config contents."""
1185	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1186	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1187	runner.invoke(cli, ["config", "set", "limits.max_ancestors", "50000"])
1188	result = runner.invoke(cli, ["config", "read", "--json"])
1189	assert result.exit_code == 0
1190	# json.loads raises on invalid JSON
1191	parsed = json.loads(result.output)
1192	assert isinstance(parsed, dict)
1193
1194
1195	# =============================================================================
1196	# muse config get — extended hardening
1197	# =============================================================================
1198
1199
1200	class TestRunGetExtended:
1201	"""Additional coverage for ``muse config get`` gaps."""
1202
1203	# ── flag aliases ──────────────────────────────────────────────────────────
1204
1205	def test_j_short_flag_emits_json(self, repo: pathlib.Path) -> None:
1206	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1207	result = runner.invoke(cli, ["config", "get", "hub.url", "-j"])
1208	assert result.exit_code == 0
1209	data = _json_get(result)
1210	assert "musehub.ai" in data["value"]
1211
1212	def test_j_same_as_json_flag(self, repo: pathlib.Path) -> None:
1213	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1214	r1 = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
1215	r2 = runner.invoke(cli, ["config", "get", "hub.url", "-j"])
1216	assert r1.exit_code == 0 and r2.exit_code == 0
1217	# Compare only stable fields — timing fields naturally differ between calls
1218	skip = {"duration_ms", "timestamp"}
1219	d1 = {k: v for k, v in _json_get(r1).items() if k not in skip}
1220	d2 = {k: v for k, v in _json_get(r2).items() if k not in skip}
1221	assert d1 == d2
1222
1223	# ── key format validation ─────────────────────────────────────────────────
1224
1225	def test_key_without_dot_exits_nonzero(self, repo: pathlib.Path) -> None:
1226	result = runner.invoke(cli, ["config", "get", "username"])
1227	assert result.exit_code != 0
1228
1229	def test_key_without_dot_shows_format_hint(self, repo: pathlib.Path) -> None:
1230	result = runner.invoke(cli, ["config", "get", "username"])
1231	assert "namespace.subkey" in result.stderr or "hub.url" in result.stderr
1232
1233	def test_key_without_dot_does_not_say_not_set(
1234	self, repo: pathlib.Path
1235	) -> None:
1236	"""Malformed key must not produce the misleading 'is not set' message."""
1237	result = runner.invoke(cli, ["config", "get", "badkey"])
1238	assert "is not set" not in result.output
1239
1240	def test_empty_key_exits_nonzero(self, repo: pathlib.Path) -> None:
1241	result = runner.invoke(cli, ["config", "get", ""])
1242	assert result.exit_code != 0
1243
1244	def test_unknown_namespace_exits_nonzero(self, repo: pathlib.Path) -> None:
1245	result = runner.invoke(cli, ["config", "get", "unknown.key"])
1246	assert result.exit_code != 0
1247
1248	# ── user.* get returns not-set (no longer in config.toml) ─────────────────
1249
1250	def test_get_user_handle_not_set(self, repo: pathlib.Path) -> None:
1251	"""user.handle is not stored in config.toml — must exit nonzero."""
1252	result = runner.invoke(cli, ["config", "get", "user.handle"])
1253	assert result.exit_code != 0
1254
1255	def test_get_user_email_not_set(self, repo: pathlib.Path) -> None:
1256	"""user.email is not stored in config.toml — must exit nonzero."""
1257	result = runner.invoke(cli, ["config", "get", "user.email"])
1258	assert result.exit_code != 0
1259
1260	def test_get_user_type_not_set(self, repo: pathlib.Path) -> None:
1261	"""user.type is not stored in config.toml — must exit nonzero."""
1262	result = runner.invoke(cli, ["config", "get", "user.type"])
1263	assert result.exit_code != 0
1264
1265	# ── all supported key namespaces ──────────────────────────────────────────
1266
1267	def test_get_hub_url(self, repo: pathlib.Path) -> None:
1268	from muse.cli.config import set_hub_url
1269	set_hub_url("https://musehub.ai", repo)
1270	result = runner.invoke(cli, ["config", "get", "hub.url"])
1271	assert result.exit_code == 0
1272	assert "musehub.ai" in result.output
1273
1274	def test_get_domain_key(self, repo: pathlib.Path) -> None:
1275	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
1276	result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"])
1277	assert result.exit_code == 0
1278	assert "960" in result.output
1279
1280	def test_get_limits_max_ancestors(self, repo: pathlib.Path) -> None:
1281	runner.invoke(cli, ["config", "set", "limits.max_ancestors", "20000"])
1282	result = runner.invoke(cli, ["config", "get", "limits.max_ancestors"])
1283	assert result.exit_code == 0
1284	assert "20000" in result.output
1285
1286	def test_get_limits_max_graph_commits(self, repo: pathlib.Path) -> None:
1287	runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "500"])
1288	result = runner.invoke(cli, ["config", "get", "limits.max_graph_commits"])
1289	assert result.exit_code == 0
1290	assert "500" in result.output
1291
1292	def test_get_limits_shard_prefix_length(self, repo: pathlib.Path) -> None:
1293	runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "4"])
1294	result = runner.invoke(cli, ["config", "get", "limits.shard_prefix_length"])
1295	assert result.exit_code == 0
1296	assert "4" in result.output
1297
1298	# ── JSON structure ────────────────────────────────────────────────────────
1299
1300	def test_json_key_field_matches_input(self, repo: pathlib.Path) -> None:
1301	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1302	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
1303	assert result.exit_code == 0
1304	data = _json_get(result)
1305	assert data["key"] == "hub.url"
1306
1307	def test_json_value_matches_text_output(self, repo: pathlib.Path) -> None:
1308	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1309	r_text = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"])
1310	r_json = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat", "--json"])
1311	assert r_text.exit_code == 0 and r_json.exit_code == 0
1312	json_value = _json_get(r_json)["value"]
1313	assert json_value in r_text.output
1314
1315	def test_json_missing_key_exits_nonzero(self, repo: pathlib.Path) -> None:
1316	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
1317	assert result.exit_code != 0
1318
1319	def test_json_stdout_clean_on_success(self, repo: pathlib.Path) -> None:
1320	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1321	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
1322	assert result.exit_code == 0
1323	# Output must be valid JSON
1324	parsed = json.loads(result.output)
1325	assert "key" in parsed and "value" in parsed
1326
1327	# ── text mode output ──────────────────────────────────────────────────────
1328
1329	def test_text_value_printed_to_stdout(self, repo: pathlib.Path) -> None:
1330	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "960"])
1331	result = runner.invoke(cli, ["config", "get", "domain.ticks_per_beat"])
1332	assert result.exit_code == 0
1333	assert "960" in result.output
1334
1335	def test_text_error_goes_to_stderr_marker(self, repo: pathlib.Path) -> None:
1336	"""Not-set error must not appear in stdout (goes to stderr)."""
1337	result = runner.invoke(cli, ["config", "get", "hub.url"])
1338	# CliRunner merges stderr — ensure exit is nonzero and message present
1339	assert result.exit_code != 0
1340	assert "not set" in result.stderr or "not" in result.stderr.lower()
1341
1342	def test_get_help_contains_key_reference(self) -> None:
1343	result = runner.invoke(cli, ["config", "get", "--help"])
1344	assert result.exit_code == 0
1345	assert "user.handle" in result.output or "hub.url" in result.output
1346
1347	def test_get_help_contains_exit_codes(self) -> None:
1348	result = runner.invoke(cli, ["config", "get", "--help"])
1349	assert result.exit_code == 0
1350	assert "Exit codes" in result.output or "exit" in result.output.lower()
1351
1352	# ── outside repo ─────────────────────────────────────────────────────────
1353
1354	def test_get_outside_repo_key_not_set(
1355	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1356	) -> None:
1357	"""Outside a repo, all keys return not-set (gracefully)."""
1358	monkeypatch.chdir(tmp_path)
1359	result = runner.invoke(cli, ["config", "get", "hub.url"])
1360	assert result.exit_code != 0
1361
1362
1363	class TestRunGetStress:
1364	"""Stress tests for ``muse config get``."""
1365
1366	def test_concurrent_get_reads(self, repo: pathlib.Path) -> None:
1367	"""Concurrent config_as_dict reads are thread-safe."""
1368	import threading
1369	from muse.cli.config import get_config_value, set_config_value
1370	set_config_value("domain.label", "StressTest", repo)
1371	errors: list[str] = []
1372
1373	def _do(idx: int) -> None:
1374	try:
1375	value = get_config_value("domain.label", repo)
1376	assert value == "StressTest"
1377	except Exception as exc:
1378	errors.append(f"Thread {idx}: {exc}")
1379
1380	threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
1381	for t in threads:
1382	t.start()
1383	for t in threads:
1384	t.join()
1385	assert errors == [], "\n".join(errors)
1386
1387	def test_all_limits_keys_readable(self, repo: pathlib.Path) -> None:
1388	"""All four limits keys must be individually gettable after set."""
1389	pairs = [
1390	("limits.max_walk_commits", "9000"),
1391	("limits.max_ancestors", "40000"),
1392	("limits.max_graph_commits", "250"),
1393	("limits.shard_prefix_length", "4"),
1394	]
1395	for key, val in pairs:
1396	runner.invoke(cli, ["config", "set", key, val])
1397	for key, expected in pairs:
1398	result = runner.invoke(cli, ["config", "get", key])
1399	assert result.exit_code == 0, f"Failed for {key}"
1400	assert expected in result.output, f"Expected {expected!r} for {key}"
1401
1402	def test_get_many_different_keys_sequentially(
1403	self, repo: pathlib.Path
1404	) -> None:
1405	"""Reading many keys in sequence must not corrupt state."""
1406	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1407	runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1408	runner.invoke(cli, ["config", "set", "domain.sample_rate", "44100"])
1409	runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "1000"])
1410	for _ in range(20):
1411	for key, expected in [
1412	("hub.url", "musehub.ai"),
1413	("domain.ticks_per_beat", "480"),
1414	("domain.sample_rate", "44100"),
1415	("limits.max_walk_commits", "1000"),
1416	]:
1417	result = runner.invoke(cli, ["config", "get", key])
1418	assert result.exit_code == 0
1419	assert expected in result.output
1420
1421
1422	# ── Extended: run_set ─────────────────────────────────────────────────────────
1423
1424
1425	class TestRunSetExtended:
1426	"""Extended hardening tests for ``muse config set``."""
1427
1428	def test_j_alias_works(self, repo: pathlib.Path) -> None:
1429	result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480", "-j"])
1430	assert result.exit_code == 0
1431	data = _json_set(result)
1432	assert data["status"] == "ok"
1433	assert data["key"] == "domain.ticks_per_beat"
1434	assert data["value"] == "480"
1435
1436	def test_key_without_dot_exits_with_format_error(self, repo: pathlib.Path) -> None:
1437	result = runner.invoke(cli, ["config", "set", "username", "Alice"])
1438	assert result.exit_code != 0
1439	assert "namespace.subkey" in result.stderr
1440
1441	def test_key_without_dot_mentions_example(self, repo: pathlib.Path) -> None:
1442	result = runner.invoke(cli, ["config", "set", "badkey", "val"])
1443	assert result.exit_code != 0
1444	assert "user.handle" in result.stderr or "hub.url" in result.stderr
1445
1446	def test_unknown_namespace_exits_nonzero(self, repo: pathlib.Path) -> None:
1447	result = runner.invoke(cli, ["config", "set", "mystery.key", "val"])
1448	assert result.exit_code != 0
1449
1450	def test_unknown_namespace_error_message_helpful(self, repo: pathlib.Path) -> None:
1451	result = runner.invoke(cli, ["config", "set", "mystery.key", "val"])
1452	assert "mystery" in result.stderr or "Unknown" in result.stderr
1453
1454	def test_user_handle_blocked(self, repo: pathlib.Path) -> None:
1455	"""user.handle is blocked — identity lives in identity.toml."""
1456	result = runner.invoke(cli, ["config", "set", "user.handle", "Bob"])
1457	assert result.exit_code != 0
1458
1459	def test_user_email_blocked(self, repo: pathlib.Path) -> None:
1460	"""user.email is blocked — identity lives in identity.toml."""
1461	result = runner.invoke(cli, ["config", "set", "user.email", "[email protected]"])
1462	assert result.exit_code != 0
1463
1464	def test_user_type_blocked(self, repo: pathlib.Path) -> None:
1465	"""user.type is blocked — identity lives in identity.toml."""
1466	result = runner.invoke(cli, ["config", "set", "user.type", "human"])
1467	assert result.exit_code != 0
1468
1469	def test_user_handle_blocked_error_mentions_auth(self, repo: pathlib.Path) -> None:
1470	"""Block message for user.* must redirect toward auth/identity commands."""
1471	result = runner.invoke(cli, ["config", "set", "user.handle", "Bob"])
1472	assert result.exit_code != 0
1473	assert "auth" in result.stderr.lower() or "identity" in result.stderr.lower()
1474
1475	def test_hub_url_https_settable(self, repo: pathlib.Path) -> None:
1476	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
1477	assert result.exit_code == 0
1478
1479	def test_hub_url_http_rejected(self, repo: pathlib.Path) -> None:
1480	result = runner.invoke(cli, ["config", "set", "hub.url", "http://musehub.ai"])
1481	assert result.exit_code != 0
1482
1483	def test_hub_unknown_subkey_rejected(self, repo: pathlib.Path) -> None:
1484	result = runner.invoke(cli, ["config", "set", "hub.secret", "val"])
1485	assert result.exit_code != 0
1486
1487	def test_domain_key_settable(self, repo: pathlib.Path) -> None:
1488	result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1489	assert result.exit_code == 0
1490
1491	def test_limits_max_ancestors_settable(self, repo: pathlib.Path) -> None:
1492	result = runner.invoke(cli, ["config", "set", "limits.max_ancestors", "25000"])
1493	assert result.exit_code == 0
1494	get_result = runner.invoke(cli, ["config", "get", "limits.max_ancestors"])
1495	assert "25000" in get_result.output
1496
1497	def test_limits_max_graph_commits_settable(self, repo: pathlib.Path) -> None:
1498	result = runner.invoke(cli, ["config", "set", "limits.max_graph_commits", "5000"])
1499	assert result.exit_code == 0
1500	get_result = runner.invoke(cli, ["config", "get", "limits.max_graph_commits"])
1501	assert "5000" in get_result.output
1502
1503	def test_limits_shard_prefix_2_valid(self, repo: pathlib.Path) -> None:
1504	result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "2"])
1505	assert result.exit_code == 0
1506
1507	def test_limits_shard_prefix_4_valid(self, repo: pathlib.Path) -> None:
1508	result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "4"])
1509	assert result.exit_code == 0
1510
1511	def test_limits_shard_prefix_1_rejected(self, repo: pathlib.Path) -> None:
1512	result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "1"])
1513	assert result.exit_code != 0
1514
1515	def test_limits_shard_prefix_3_rejected(self, repo: pathlib.Path) -> None:
1516	result = runner.invoke(cli, ["config", "set", "limits.shard_prefix_length", "3"])
1517	assert result.exit_code != 0
1518
1519	def test_limits_negative_rejected(self, repo: pathlib.Path) -> None:
1520	result = runner.invoke(cli, ["config", "set", "limits.max_walk_commits", "-5"])
1521	assert result.exit_code != 0
1522
1523	def test_blocked_auth_shows_redirect(self, repo: pathlib.Path) -> None:
1524	result = runner.invoke(cli, ["config", "set", "auth.token", "secret"])
1525	assert result.exit_code != 0
1526	assert "auth" in result.stderr.lower() or "login" in result.stderr.lower()
1527
1528	def test_blocked_remotes_shows_redirect(self, repo: pathlib.Path) -> None:
1529	result = runner.invoke(cli, ["config", "set", "remotes.origin", "url"])
1530	assert result.exit_code != 0
1531	assert "remote" in result.stderr.lower()
1532
1533	def test_success_json_status_ok(self, repo: pathlib.Path) -> None:
1534	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
1535	data = _json_set(result)
1536	assert data["status"] == "ok"
1537
1538	def test_success_json_stdout_only_has_json(self, repo: pathlib.Path) -> None:
1539	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
1540	assert result.exit_code == 0
1541	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
1542	assert len(json_lines) >= 1
1543
1544	def test_success_text_goes_to_output(self, repo: pathlib.Path) -> None:
1545	result = runner.invoke(cli, ["config", "set", "domain.ticks_per_beat", "480"])
1546	assert result.exit_code == 0
1547	assert "480" in result.stderr
1548
1549	def test_help_contains_settable_namespaces(self, repo: pathlib.Path) -> None:
1550	result = runner.invoke(cli, ["config", "set", "--help"])
1551	assert "hub.url" in result.output
1552	assert "domain" in result.output
1553	assert "limits" in result.output
1554
1555	def test_help_contains_blocked_namespaces(self, repo: pathlib.Path) -> None:
1556	result = runner.invoke(cli, ["config", "set", "--help"])
1557	assert "auth" in result.output
1558	assert "remotes" in result.output
1559
1560	def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1561	result = runner.invoke(cli, ["config", "set", "--help"])
1562	assert "Exit" in result.output or "exit" in result.output
1563
1564
1565	# ── Security: run_set ─────────────────────────────────────────────────────────
1566
1567
1568	class TestRunSetSecurity:
1569	"""Security-focused tests for ``muse config set``."""
1570
1571	def test_ansi_in_key_sanitized_in_error(self, repo: pathlib.Path) -> None:
1572	result = runner.invoke(cli, ["config", "set", "domain.\x1b[31mmalicious\x1b[0m\nkey", "val"])
1573	assert result.exit_code != 0
1574	assert "\x1b[" not in result.output
1575
1576	def test_ansi_in_value_sanitized_in_text_success(self, repo: pathlib.Path) -> None:
1577	result = runner.invoke(cli, ["config", "set", "domain.label", "\x1b[31mRed\x1b[0m"])
1578	assert result.exit_code == 0
1579	assert "\x1b[" not in result.output
1580
1581	def test_null_byte_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1582	result = runner.invoke(cli, ["config", "set", "domain.malicious\x00key", "val"])
1583	assert result.exit_code != 0
1584
1585	def test_bracket_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1586	result = runner.invoke(cli, ["config", "set", "domain.x][malicious", "val"])
1587	assert result.exit_code != 0
1588
1589	def test_equals_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1590	result = runner.invoke(cli, ["config", "set", "domain.key=malicious", "val"])
1591	assert result.exit_code != 0
1592
1593	def test_newline_in_domain_key_rejected(self, repo: pathlib.Path) -> None:
1594	result = runner.invoke(cli, ["config", "set", "domain.malicious\nkey", "val"])
1595	assert result.exit_code != 0
1596
1597	def test_key_without_dot_shows_format_hint_not_generic_error(self, repo: pathlib.Path) -> None:
1598	"""Ensures format error, not a generic 'unknown namespace' message."""
1599	result = runner.invoke(cli, ["config", "set", "nodot", "val"])
1600	assert result.exit_code != 0
1601	assert "namespace.subkey" in result.stderr
1602
1603	def test_hub_http_blocked_not_stored(self, repo: pathlib.Path) -> None:
1604	runner.invoke(cli, ["config", "set", "hub.url", "http://attacker.example.com"])
1605	result = runner.invoke(cli, ["config", "get", "hub.url"])
1606	assert result.exit_code != 0 or "attacker.example.com" not in result.output
1607
1608	def test_user_namespace_not_stored_on_block(self, repo: pathlib.Path) -> None:
1609	"""Blocked user.* write must not write anything to config.toml."""
1610	runner.invoke(cli, ["config", "set", "user.handle", "ShouldNotStore"])
1611	raw = (config_toml_path(repo)).read_text()
1612	assert "ShouldNotStore" not in raw
1613	assert "[user]" not in raw
1614
1615
1616	# ── Stress: run_set ───────────────────────────────────────────────────────────
1617
1618
1619	class TestRunSetStress:
1620	"""Concurrency and volume tests for ``muse config set``."""
1621
1622	def test_concurrent_set_to_different_repos_no_corruption(
1623	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1624	) -> None:
1625	"""8 threads writing to separate repos must each produce correct values."""
1626	import threading
1627	from muse._version import __version__
1628	from muse.cli.config import get_config_value, set_config_value
1629
1630	errors: list[str] = []
1631
1632	def _worker(idx: int) -> None:
1633	repo_path = tmp_path / f"repo_{idx}"
1634	dot_muse = muse_dir(repo_path)
1635	for sub in ("refs/heads", "objects", "commits", "snapshots"):
1636	(dot_muse / sub).mkdir(parents=True, exist_ok=True)
1637	(dot_muse / "repo.json").write_text(
1638	json.dumps({"repo_id": f"repo-{idx}", "schema_version": __version__, "domain": "code"})
1639	)
1640	(dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
1641	(dot_muse / "config.toml").write_text("")
1642	expected = f"val_{idx}"
1643	set_config_value("domain.label", expected, repo_path)
1644	got = get_config_value("domain.label", repo_path)
1645	if got != expected:
1646	errors.append(f"repo_{idx}: expected {expected!r}, got {got!r}")
1647
1648	threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
1649	for t in threads:
1650	t.start()
1651	for t in threads:
1652	t.join()
1653	assert errors == [], "\n".join(errors)
1654
1655	def test_all_four_limits_keys_set_round_trip(self, repo: pathlib.Path) -> None:
1656	"""All four limits keys written then read back."""
1657	from muse.cli.config import get_config_value, set_config_value
1658
1659	pairs = [
1660	("limits.max_walk_commits", "10000"),
1661	("limits.max_ancestors", "5000"),
1662	("limits.max_graph_commits", "2500"),
1663	("limits.shard_prefix_length", "4"),
1664	]
1665	for key, val in pairs:
1666	set_config_value(key, val, repo)
1667	for key, expected in pairs:
1668	got = get_config_value(key, repo)
1669	assert got == expected, f"{key}: expected {expected!r}, got {got!r}"
1670
1671	def test_20_sequential_domain_sets_no_state_corruption(self, repo: pathlib.Path) -> None:
1672	"""Writing 20 different domain.label values sequentially — last write wins."""
1673	from muse.cli.config import get_config_value, set_config_value
1674
1675	for i in range(20):
1676	set_config_value("domain.label", f"val_{i}", repo)
1677	got = get_config_value("domain.label", repo)
1678	assert got == "val_19"
1679
1680	def test_user_set_always_raises_not_written(self, repo: pathlib.Path) -> None:
1681	"""20 sequential user.* writes must all raise ValueError, nothing stored."""
1682	from muse.cli.config import set_config_value
1683
1684	for i in range(20):
1685	with pytest.raises(ValueError):
1686	set_config_value("user.handle", f"user_{i}", repo)
1687	raw = (config_toml_path(repo)).read_text()
1688	assert "[user]" not in raw
1689
1690
1691	# ── Extended: run_edit ────────────────────────────────────────────────────────
1692
1693
1694	class TestRunEditExtended:
1695	"""Extended hardening tests for ``muse config edit``."""
1696
1697	def test_visual_takes_precedence_over_editor(
1698	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1699	) -> None:
1700	"""$VISUAL must be used when both $VISUAL and $EDITOR are set."""
1701	calls: list[list[str]] = []
1702
1703	import subprocess as _sp
1704
1705	real_run = _sp.run
1706
1707	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1708	calls.append(cmd)
1709	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1710
1711	monkeypatch.setattr(_sp, "run", _fake_run)
1712	monkeypatch.setenv("VISUAL", "visual-editor")
1713	monkeypatch.setenv("EDITOR", "editor-fallback")
1714	runner.invoke(cli, ["config", "edit"])
1715	assert calls and calls[0][0] == "visual-editor"
1716
1717	def test_editor_used_when_visual_unset(
1718	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1719	) -> None:
1720	calls: list[list[str]] = []
1721
1722	import subprocess as _sp
1723
1724	real_run = _sp.run
1725
1726	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1727	calls.append(cmd)
1728	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1729
1730	monkeypatch.setattr(_sp, "run", _fake_run)
1731	monkeypatch.delenv("VISUAL", raising=False)
1732	monkeypatch.setenv("EDITOR", "my-editor")
1733	runner.invoke(cli, ["config", "edit"])
1734	assert calls and calls[0][0] == "my-editor"
1735
1736	def test_vi_fallback_when_both_unset(
1737	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1738	) -> None:
1739	calls: list[list[str]] = []
1740
1741	import subprocess as _sp
1742
1743	real_run = _sp.run
1744
1745	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1746	calls.append(cmd)
1747	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1748
1749	monkeypatch.setattr(_sp, "run", _fake_run)
1750	monkeypatch.delenv("VISUAL", raising=False)
1751	monkeypatch.delenv("EDITOR", raising=False)
1752	runner.invoke(cli, ["config", "edit"])
1753	assert calls and calls[0][0] == "vi"
1754
1755	def test_multiword_editor_split_correctly(
1756	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1757	) -> None:
1758	"""EDITOR='code --wait' must be split to ['code', '--wait', path]."""
1759	calls: list[list[str]] = []
1760
1761	import subprocess as _sp
1762
1763	real_run = _sp.run
1764
1765	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1766	calls.append(cmd)
1767	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1768
1769	monkeypatch.setattr(_sp, "run", _fake_run)
1770	monkeypatch.setenv("EDITOR", "code --wait")
1771	monkeypatch.delenv("VISUAL", raising=False)
1772	runner.invoke(cli, ["config", "edit"])
1773	assert calls and calls[0][:2] == ["code", "--wait"]
1774
1775	def test_multiword_editor_passes_config_path(
1776	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1777	) -> None:
1778	"""Multi-word editor: config path must be the final argument."""
1779	calls: list[list[str]] = []
1780
1781	import subprocess as _sp
1782
1783	real_run = _sp.run
1784
1785	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1786	calls.append(cmd)
1787	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1788
1789	monkeypatch.setattr(_sp, "run", _fake_run)
1790	monkeypatch.setenv("EDITOR", "emacs -nw")
1791	monkeypatch.delenv("VISUAL", raising=False)
1792	runner.invoke(cli, ["config", "edit"])
1793	assert calls and "config.toml" in calls[0][-1]
1794
1795	def test_auto_create_config_when_missing(
1796	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1797	) -> None:
1798	(config_toml_path(repo)).unlink()
1799	monkeypatch.setenv("EDITOR", "true")
1800	monkeypatch.delenv("VISUAL", raising=False)
1801	assert not (config_toml_path(repo)).exists()
1802	result = runner.invoke(cli, ["config", "edit"])
1803	assert result.exit_code == 0
1804	assert (config_toml_path(repo)).exists()
1805
1806	def test_auto_create_info_message_on_stderr(
1807	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1808	) -> None:
1809	(config_toml_path(repo)).unlink()
1810	monkeypatch.setenv("EDITOR", "true")
1811	monkeypatch.delenv("VISUAL", raising=False)
1812	result = runner.invoke(cli, ["config", "edit"])
1813	assert "Created" in result.stderr or "config.toml" in result.stderr
1814
1815	def test_editor_invoked_as_list_not_shell(
1816	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1817	) -> None:
1818	"""subprocess.run must receive a list, never shell=True."""
1819	captured: MsgpackDict = {}
1820
1821	import subprocess as _sp
1822
1823	def _fake_run(cmd: list[str] \| str, **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1824	captured["cmd"] = cmd
1825	captured["shell"] = kwargs.get("shell", False)
1826	from subprocess import CompletedProcess
1827	return CompletedProcess([], 0)
1828
1829	monkeypatch.setattr(_sp, "run", _fake_run)
1830	monkeypatch.setenv("EDITOR", "true")
1831	monkeypatch.delenv("VISUAL", raising=False)
1832	runner.invoke(cli, ["config", "edit"])
1833	assert isinstance(captured.get("cmd"), list)
1834	assert not captured.get("shell")
1835
1836	def test_editor_nonzero_exit_exits_nonzero(
1837	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1838	) -> None:
1839	monkeypatch.setenv("EDITOR", "false") # /bin/false always exits 1
1840	monkeypatch.delenv("VISUAL", raising=False)
1841	result = runner.invoke(cli, ["config", "edit"])
1842	assert result.exit_code != 0
1843
1844	def test_editor_nonzero_exit_message_contains_code(
1845	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1846	) -> None:
1847	monkeypatch.setenv("EDITOR", "false")
1848	monkeypatch.delenv("VISUAL", raising=False)
1849	result = runner.invoke(cli, ["config", "edit"])
1850	assert result.exit_code != 0
1851	# Message should mention the exit code number
1852	assert any(ch.isdigit() for ch in result.stderr)
1853
1854	def test_outside_repo_exits_nonzero(
1855	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1856	) -> None:
1857	monkeypatch.chdir(tmp_path)
1858	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1859	result = runner.invoke(cli, ["config", "edit"])
1860	assert result.exit_code != 0
1861
1862	def test_outside_repo_error_message(
1863	self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1864	) -> None:
1865	monkeypatch.chdir(tmp_path)
1866	monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1867	result = runner.invoke(cli, ["config", "edit"])
1868	assert "repository" in result.stderr.lower() or "repo" in result.stderr.lower()
1869
1870	def test_help_mentions_visual(self, repo: pathlib.Path) -> None:
1871	result = runner.invoke(cli, ["config", "edit", "--help"])
1872	assert "VISUAL" in result.output
1873
1874	def test_help_mentions_editor(self, repo: pathlib.Path) -> None:
1875	result = runner.invoke(cli, ["config", "edit", "--help"])
1876	assert "EDITOR" in result.output
1877
1878	def test_help_mentions_agent_alternative(self, repo: pathlib.Path) -> None:
1879	result = runner.invoke(cli, ["config", "edit", "--help"])
1880	assert "muse config set" in result.output or "agent" in result.output
1881
1882	def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1883	result = runner.invoke(cli, ["config", "edit", "--help"])
1884	assert "Exit" in result.output or "exit" in result.output
1885
1886	def test_config_path_passed_to_editor(
1887	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1888	) -> None:
1889	"""Editor must receive the config.toml path as its argument."""
1890	calls: list[list[str]] = []
1891
1892	import subprocess as _sp
1893
1894	real_run = _sp.run
1895
1896	def _fake_run(cmd: list[str], **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1897	calls.append(cmd)
1898	return real_run(["true"], **{k: v for k, v in kwargs.items()})
1899
1900	monkeypatch.setattr(_sp, "run", _fake_run)
1901	monkeypatch.setenv("EDITOR", "my-editor")
1902	monkeypatch.delenv("VISUAL", raising=False)
1903	runner.invoke(cli, ["config", "edit"])
1904	assert calls
1905	assert str(config_toml_path(repo)) in calls[0]
1906
1907
1908	# ── Security: run_edit ────────────────────────────────────────────────────────
1909
1910
1911	class TestRunEditSecurity:
1912	"""Security-focused tests for ``muse config edit``."""
1913
1914	def test_ansi_in_editor_env_sanitized_in_error(
1915	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1916	) -> None:
1917	monkeypatch.setenv("EDITOR", "\x1b[31mmalicious\x1b[0m-editor")
1918	monkeypatch.delenv("VISUAL", raising=False)
1919	result = runner.invoke(cli, ["config", "edit"])
1920	assert result.exit_code != 0
1921	assert "\x1b[" not in result.output
1922
1923	def test_editor_invoked_without_shell(
1924	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1925	) -> None:
1926	"""Verify shell=True is never passed to subprocess.run."""
1927	captured: MsgpackDict = {}
1928
1929	import subprocess as _sp
1930
1931	def _fake_run(cmd: list[str] \| str, **kwargs: str \| bool \| int \| None) -> _sp.CompletedProcess[bytes]:
1932	captured["shell"] = kwargs.get("shell", False)
1933	from subprocess import CompletedProcess
1934	return CompletedProcess([], 0)
1935
1936	monkeypatch.setattr(_sp, "run", _fake_run)
1937	monkeypatch.setenv("EDITOR", "true")
1938	monkeypatch.delenv("VISUAL", raising=False)
1939	runner.invoke(cli, ["config", "edit"])
1940	assert not captured.get("shell")
1941
1942	def test_malformed_editor_command_exits_gracefully(
1943	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1944	) -> None:
1945	"""An unparseable $EDITOR value must exit cleanly, not crash."""
1946	# shlex.split raises ValueError on unmatched quotes
1947	monkeypatch.setenv("EDITOR", "editor 'unclosed quote")
1948	monkeypatch.delenv("VISUAL", raising=False)
1949	result = runner.invoke(cli, ["config", "edit"])
1950	assert result.exit_code != 0
1951
1952	def test_shell_metacharacters_in_editor_not_shell_executed(
1953	self, repo: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1954	) -> None:
1955	"""$EDITOR with shell metacharacters must not trigger shell execution.
1956
1957	shlex.split turns 'true; echo injected' into ['true;', 'echo', 'injected'].
1958	subprocess.run receives a list (never shell=True), so it tries to exec a
1959	binary literally named 'true;' — which doesn't exist. FileNotFoundError
1960	fires; no shell command is ever evaluated.
1961	"""
1962	monkeypatch.setenv("EDITOR", "true; echo injected")
1963	monkeypatch.delenv("VISUAL", raising=False)
1964	result = runner.invoke(cli, ["config", "edit"])
1965	# Binary 'true;' doesn't exist → non-zero exit; no shell evaluation.
1966	assert result.exit_code != 0
1967	# Error message quotes the editor string, not the result of shell execution.
1968	assert "Editor not found" in result.stderr
1969
1970
1971	# =============================================================================
1972	# Supercharge — duration_ms + exit_code in all JSON paths
1973	# =============================================================================
1974
1975	_GET_FULL_KEYS = frozenset({"key", "value", "duration_ms", "exit_code"})
1976	_SET_FULL_KEYS = frozenset({"status", "key", "value", "duration_ms", "exit_code"})
1977
1978
1979	class TestElapsedSeconds:
1980	"""duration_ms must appear in every JSON output path."""
1981
1982	def test_read_json_has_elapsed(self, repo: pathlib.Path) -> None:
1983	result = runner.invoke(cli, ["config", "read", "--json"])
1984	assert result.exit_code == 0
1985	data = json.loads(result.output)
1986	assert "duration_ms" in data
1987
1988	def test_read_elapsed_is_float(self, repo: pathlib.Path) -> None:
1989	result = runner.invoke(cli, ["config", "read", "--json"])
1990	assert result.exit_code == 0
1991	data = json.loads(result.output)
1992	assert isinstance(data["duration_ms"], float)
1993
1994	def test_read_elapsed_non_negative(self, repo: pathlib.Path) -> None:
1995	result = runner.invoke(cli, ["config", "read", "--json"])
1996	assert result.exit_code == 0
1997	data = json.loads(result.output)
1998	assert data["duration_ms"] >= 0.0
1999
2000	def test_get_json_has_elapsed(self, repo: pathlib.Path) -> None:
2001	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2002	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2003	assert result.exit_code == 0
2004	data = json.loads(result.output)
2005	assert "duration_ms" in data
2006
2007	def test_get_elapsed_is_float(self, repo: pathlib.Path) -> None:
2008	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2009	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2010	assert result.exit_code == 0
2011	data = json.loads(result.output)
2012	assert isinstance(data["duration_ms"], float)
2013
2014	def test_get_elapsed_non_negative(self, repo: pathlib.Path) -> None:
2015	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2016	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2017	assert result.exit_code == 0
2018	data = json.loads(result.output)
2019	assert data["duration_ms"] >= 0.0
2020
2021	def test_set_json_has_elapsed(self, repo: pathlib.Path) -> None:
2022	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2023	assert result.exit_code == 0
2024	data = json.loads(result.output)
2025	assert "duration_ms" in data
2026
2027	def test_set_elapsed_is_float(self, repo: pathlib.Path) -> None:
2028	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2029	assert result.exit_code == 0
2030	data = json.loads(result.output)
2031	assert isinstance(data["duration_ms"], float)
2032
2033	def test_set_elapsed_non_negative(self, repo: pathlib.Path) -> None:
2034	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2035	assert result.exit_code == 0
2036	data = json.loads(result.output)
2037	assert data["duration_ms"] >= 0.0
2038
2039
2040	class TestExitCode:
2041	"""exit_code must appear in every JSON output path and mirror process exit."""
2042
2043	def test_read_json_has_exit_code(self, repo: pathlib.Path) -> None:
2044	result = runner.invoke(cli, ["config", "read", "--json"])
2045	assert result.exit_code == 0
2046	data = json.loads(result.output)
2047	assert "exit_code" in data
2048
2049	def test_read_exit_code_zero(self, repo: pathlib.Path) -> None:
2050	result = runner.invoke(cli, ["config", "read", "--json"])
2051	assert result.exit_code == 0
2052	data = json.loads(result.output)
2053	assert data["exit_code"] == 0
2054
2055	def test_get_json_has_exit_code(self, repo: pathlib.Path) -> None:
2056	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2057	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2058	assert result.exit_code == 0
2059	data = json.loads(result.output)
2060	assert "exit_code" in data
2061
2062	def test_get_exit_code_zero_on_success(self, repo: pathlib.Path) -> None:
2063	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2064	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2065	assert result.exit_code == 0
2066	data = json.loads(result.output)
2067	assert data["exit_code"] == 0
2068
2069	def test_set_json_has_exit_code(self, repo: pathlib.Path) -> None:
2070	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2071	assert result.exit_code == 0
2072	data = json.loads(result.output)
2073	assert "exit_code" in data
2074
2075	def test_set_exit_code_zero_on_success(self, repo: pathlib.Path) -> None:
2076	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2077	assert result.exit_code == 0
2078	data = json.loads(result.output)
2079	assert data["exit_code"] == 0
2080
2081
2082	class TestJsonSchemaComplete:
2083	"""Every JSON output path must carry the full schema — no follow-up reads needed."""
2084
2085	def test_read_full_keys(self, repo: pathlib.Path) -> None:
2086	result = runner.invoke(cli, ["config", "read", "--json"])
2087	assert result.exit_code == 0
2088	data = json.loads(result.output)
2089	assert "duration_ms" in data and "exit_code" in data
2090
2091	def test_get_full_keys(self, repo: pathlib.Path) -> None:
2092	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2093	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2094	assert result.exit_code == 0
2095	data = json.loads(result.output)
2096	missing = _GET_FULL_KEYS - set(data.keys())
2097	assert not missing, f"Missing keys: {missing}"
2098
2099	def test_set_full_keys(self, repo: pathlib.Path) -> None:
2100	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2101	assert result.exit_code == 0
2102	data = json.loads(result.output)
2103	missing = _SET_FULL_KEYS - set(data.keys())
2104	assert not missing, f"Missing keys: {missing}"
2105
2106	def test_get_output_is_single_line_json(self, repo: pathlib.Path) -> None:
2107	runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai"])
2108	result = runner.invoke(cli, ["config", "get", "hub.url", "--json"])
2109	assert result.exit_code == 0
2110	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2111	assert len(json_lines) == 1
2112	json.loads(json_lines[0])
2113
2114	def test_set_output_is_single_line_json(self, repo: pathlib.Path) -> None:
2115	result = runner.invoke(cli, ["config", "set", "hub.url", "https://musehub.ai", "--json"])
2116	assert result.exit_code == 0
2117	json_lines = [l for l in result.output.splitlines() if l.strip().startswith("{")]
2118	assert len(json_lines) == 1
2119	json.loads(json_lines[0])

test_cmd_config_hardening.py file-level

`test_cmd_config_hardening.py` file-level