gabriel / muse public
test_migrate_hub_scoping.py python
978 lines 42.2 KB
Raw
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ breaking 8 hours ago
1 """TDD tests for ``muse migrate hub-scoping`` (musehub#221).
2
3 Covers:
4
5 1. Core library (muse.core.hub_scoping_migration)
6 - Pre-Phase-2 path detection (identity domain only)
7 - Old -> new path mapping, hub-scoped
8 - Key re-derivation produces a different (correct) fingerprint per hub
9 - Identity file scanning
10 2. Dry-run plan (no writes, no hub calls)
11 3. Live run (hub called, identity map mutated)
12 4. CLI smoke (muse migrate hub-scoping --dry-run / --no-register)
13 5. Security adversarial inputs and boundary conditions
14
15 Background
16 ----------
17 Prior to Phase 2, the identity key at rotation index 0 was bit-for-bit
18 identical regardless of which hub it was registered with. Phase 2 inserts a
19 hardened ``hub'`` level between ``role'`` and ``index'``. Users with keys
20 derived before Phase 2 must re-derive at the new, hub-scoped path and
21 re-register with the affected hub.
22 """
23
24 from __future__ import annotations
25
26 import json
27 import pathlib
28 from collections.abc import Mapping
29 from unittest.mock import MagicMock
30
31 import pytest
32
33 from muse.core.hdkeys import (
34 DOMAIN_IDENTITY,
35 DOMAIN_CODE,
36 ENTITY_AGENT,
37 ROLE_ATTEST,
38 hub_index,
39 muse_path,
40 )
41 from muse.core.paths import muse_dir
42 from muse.core.slip010 import MUSE_PURPOSE
43
44 FAKE_MNEMONIC = (
45 "abandon abandon abandon abandon abandon abandon "
46 "abandon abandon abandon abandon abandon about"
47 )
48
49 # ---------------------------------------------------------------------------
50 # Helpers
51 # ---------------------------------------------------------------------------
52
53
54 def _pre_scoping_path(entity_type: int = 0, entity_id: int = 0, role: int = 0, index: int = 0) -> str:
55 """Build a pre-Phase-2 six-level identity path (no hub segment)."""
56 return muse_path(DOMAIN_IDENTITY, entity_type, entity_id, role, index)
57
58
59 # ============================================================================
60 # 1. Core: pre-Phase-2 path detection
61 # ============================================================================
62
63
64 class TestIsPreHubScopingHdPath:
65 def test_six_level_identity_path_is_pre_scoping(self) -> None:
66 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
67 assert is_pre_hub_scoping_hd_path(_pre_scoping_path()) is True
68
69 def test_seven_level_hub_scoped_path_is_not_pre_scoping(self) -> None:
70 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
71 hub = hub_index("musehub.ai")
72 scoped = muse_path(DOMAIN_IDENTITY, hub=hub)
73 assert is_pre_hub_scoping_hd_path(scoped) is False
74
75 def test_non_identity_domain_six_level_path_is_not_flagged(self) -> None:
76 """Code/music/etc. domains never had hub scoping — not part of this migration."""
77 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
78 assert is_pre_hub_scoping_hd_path(muse_path(DOMAIN_CODE)) is False
79
80 def test_empty_string_is_not_pre_scoping(self) -> None:
81 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
82 assert is_pre_hub_scoping_hd_path("") is False
83
84 def test_non_muse_purpose_is_not_pre_scoping(self) -> None:
85 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
86 assert is_pre_hub_scoping_hd_path("m/44'/0'/0'/0'/0'/0'") is False
87
88 def test_agent_pre_scoping_path_is_flagged(self) -> None:
89 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
90 assert is_pre_hub_scoping_hd_path(
91 _pre_scoping_path(entity_type=ENTITY_AGENT, entity_id=3)
92 ) is True
93
94
95 # ============================================================================
96 # 2. Core: old -> new path mapping
97 # ============================================================================
98
99
100 class TestNewPathForPreHubScoping:
101 def test_inserts_hub_segment_before_index(self) -> None:
102 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
103 old = _pre_scoping_path()
104 new = new_path_for_pre_hub_scoping(old, "musehub.ai")
105 expected_hub = hub_index("musehub.ai")
106 assert new == muse_path(DOMAIN_IDENTITY, hub=expected_hub)
107
108 def test_different_hubs_produce_different_paths(self) -> None:
109 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
110 old = _pre_scoping_path()
111 a = new_path_for_pre_hub_scoping(old, "musehub.ai")
112 b = new_path_for_pre_hub_scoping(old, "staging.musehub.ai")
113 assert a != b
114
115 def test_preserves_entity_type_and_id(self) -> None:
116 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
117 old = _pre_scoping_path(entity_type=ENTITY_AGENT, entity_id=5)
118 new = new_path_for_pre_hub_scoping(old, "musehub.ai")
119 parts = new.split("/")
120 assert parts[3] == f"{ENTITY_AGENT}'"
121 assert parts[4] == "5'"
122
123 def test_preserves_role_and_index(self) -> None:
124 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
125 old = _pre_scoping_path(role=ROLE_ATTEST, index=2)
126 new = new_path_for_pre_hub_scoping(old, "musehub.ai")
127 parts = new.split("/")
128 assert parts[5] == f"{ROLE_ATTEST}'"
129 assert parts[-1] == "2'"
130
131 def test_non_identity_domain_raises(self) -> None:
132 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
133 with pytest.raises(ValueError, match="Not an identity-domain path"):
134 new_path_for_pre_hub_scoping(muse_path(DOMAIN_CODE), "musehub.ai")
135
136 def test_already_hub_scoped_path_raises(self) -> None:
137 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
138 scoped = muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai"))
139 with pytest.raises(ValueError):
140 new_path_for_pre_hub_scoping(scoped, "musehub.ai")
141
142 def test_output_has_seven_hardened_segments(self) -> None:
143 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
144 new = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai")
145 assert new.startswith(f"m/{MUSE_PURPOSE}'")
146 parts = new.split("/")[1:]
147 assert len(parts) == 7
148 assert all(p.endswith("'") for p in parts)
149
150
151 # ============================================================================
152 # 3. Core: key re-derivation
153 # ============================================================================
154
155
156 class TestDeriveFingerprintAtHubScopedPath:
157 def test_old_and_new_fingerprints_differ(self) -> None:
158 from muse.core.hub_scoping_migration import (
159 derive_fingerprint_at_hub_scoped_path,
160 new_path_for_pre_hub_scoping,
161 )
162 from muse.core.domain_migration import derive_fingerprint_at_path
163 from muse.core.bip39 import mnemonic_to_seed
164
165 seed = mnemonic_to_seed(FAKE_MNEMONIC)
166 old_fp = derive_fingerprint_at_path(seed, _pre_scoping_path())
167 new_path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai")
168 new_fp = derive_fingerprint_at_hub_scoped_path(seed, new_path)
169 assert old_fp != new_fp
170
171 def test_different_hubs_produce_different_fingerprints(self) -> None:
172 from muse.core.hub_scoping_migration import (
173 derive_fingerprint_at_hub_scoped_path,
174 new_path_for_pre_hub_scoping,
175 )
176 from muse.core.bip39 import mnemonic_to_seed
177
178 seed = mnemonic_to_seed(FAKE_MNEMONIC)
179 path_a = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai")
180 path_b = new_path_for_pre_hub_scoping(_pre_scoping_path(), "staging.musehub.ai")
181 fp_a = derive_fingerprint_at_hub_scoped_path(seed, path_a)
182 fp_b = derive_fingerprint_at_hub_scoped_path(seed, path_b)
183 assert fp_a != fp_b
184
185 def test_deterministic(self) -> None:
186 from muse.core.hub_scoping_migration import (
187 derive_fingerprint_at_hub_scoped_path,
188 new_path_for_pre_hub_scoping,
189 )
190 from muse.core.bip39 import mnemonic_to_seed
191
192 seed = mnemonic_to_seed(FAKE_MNEMONIC)
193 path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai")
194 fp1 = derive_fingerprint_at_hub_scoped_path(seed, path)
195 fp2 = derive_fingerprint_at_hub_scoped_path(seed, path)
196 assert fp1 == fp2
197
198 def test_matches_direct_derive_identity_key(self) -> None:
199 from muse.core.hub_scoping_migration import derive_fingerprint_at_hub_scoped_path
200 from muse.core.bip39 import mnemonic_to_seed
201 from muse.core.keypair import derive_hd_public_info
202
203 seed = mnemonic_to_seed(FAKE_MNEMONIC)
204 hub = hub_index("musehub.ai")
205 _, expected_fp = derive_hd_public_info(seed, hub=hub)
206 actual_fp = derive_fingerprint_at_hub_scoped_path(seed, muse_path(DOMAIN_IDENTITY, hub=hub))
207 assert actual_fp == expected_fp
208
209 def test_fingerprint_is_sha256_prefixed(self) -> None:
210 from muse.core.hub_scoping_migration import (
211 derive_fingerprint_at_hub_scoped_path,
212 new_path_for_pre_hub_scoping,
213 )
214 from muse.core.bip39 import mnemonic_to_seed
215
216 seed = mnemonic_to_seed(FAKE_MNEMONIC)
217 path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai")
218 fp = derive_fingerprint_at_hub_scoped_path(seed, path)
219 assert fp.startswith("sha256:")
220 assert len(fp) == 71
221
222 def test_rejects_six_level_path(self) -> None:
223 from muse.core.hub_scoping_migration import derive_fingerprint_at_hub_scoped_path
224 from muse.core.bip39 import mnemonic_to_seed
225 seed = mnemonic_to_seed(FAKE_MNEMONIC)
226 with pytest.raises(ValueError, match="Cannot parse"):
227 derive_fingerprint_at_hub_scoped_path(seed, _pre_scoping_path())
228
229
230 # ============================================================================
231 # 4. Core: scanning identity map
232 # ============================================================================
233
234
235 class TestScanForPreHubScoping:
236 def test_finds_pre_scoping_entry(self) -> None:
237 from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping
238 identity_map = {
239 "musehub.ai": {"type": "human", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64},
240 }
241 assert "musehub.ai" in scan_for_pre_hub_scoping(identity_map)
242
243 def test_ignores_already_scoped_entry(self) -> None:
244 from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping
245 identity_map = {
246 "musehub.ai": {
247 "type": "human",
248 "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")),
249 "fingerprint": "b" * 64,
250 },
251 }
252 assert scan_for_pre_hub_scoping(identity_map) == []
253
254 def test_ignores_non_identity_domain_entry(self) -> None:
255 from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping
256 identity_map = {"musehub.ai": {"hd_path": muse_path(DOMAIN_CODE), "fingerprint": "c" * 64}}
257 assert scan_for_pre_hub_scoping(identity_map) == []
258
259 def test_finds_multiple_hubs(self) -> None:
260 from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping
261 identity_map = {
262 "musehub.ai": {"hd_path": _pre_scoping_path(), "fingerprint": "a" * 64},
263 "staging.musehub.ai": {"hd_path": _pre_scoping_path(), "fingerprint": "b" * 64},
264 }
265 assert set(scan_for_pre_hub_scoping(identity_map)) == {"musehub.ai", "staging.musehub.ai"}
266
267 def test_empty_map_returns_empty(self) -> None:
268 from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping
269 assert scan_for_pre_hub_scoping({}) == []
270
271
272 # ============================================================================
273 # 5. Dry-run and live run
274 # ============================================================================
275
276
277 class TestDryRun:
278 def test_dry_run_returns_plans_without_registering(self) -> None:
279 from muse.core.hub_scoping_migration import run_migration
280 from muse.core.bip39 import mnemonic_to_seed
281
282 identity_map = {
283 "musehub.ai": {
284 "type": "human", "handle": "gabriel",
285 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
286 }
287 }
288 seed = mnemonic_to_seed(FAKE_MNEMONIC)
289 hub_register = MagicMock()
290
291 result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=True)
292
293 hub_register.assert_not_called()
294 assert len(result) == 1
295 assert result[0].hub_registered is False
296 assert result[0].new_fingerprint != result[0].old_fingerprint
297
298 def test_dry_run_does_not_mutate_identity_map(self) -> None:
299 from muse.core.hub_scoping_migration import run_migration
300 from muse.core.bip39 import mnemonic_to_seed
301
302 identity_map = {
303 "musehub.ai": {"type": "human", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64},
304 }
305 seed = mnemonic_to_seed(FAKE_MNEMONIC)
306 run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True)
307 assert identity_map["musehub.ai"]["hd_path"] == _pre_scoping_path()
308 assert identity_map["musehub.ai"]["fingerprint"] == "a" * 64
309
310
311 class TestLiveMigration:
312 def test_live_run_calls_hub_register_and_updates_map(self) -> None:
313 from muse.core.hub_scoping_migration import run_migration
314 from muse.core.bip39 import mnemonic_to_seed
315
316 identity_map = {
317 "musehub.ai": {
318 "type": "human", "handle": "gabriel",
319 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
320 }
321 }
322 seed = mnemonic_to_seed(FAKE_MNEMONIC)
323 hub_register = MagicMock(return_value=True)
324
325 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
326
327 hub_register.assert_called_once()
328 assert results[0].hub_registered is True
329 assert identity_map["musehub.ai"]["hd_path"] == results[0].new_hd_path
330 assert identity_map["musehub.ai"]["fingerprint"] == results[0].new_fingerprint
331
332 def test_live_run_skips_already_scoped_entries(self) -> None:
333 from muse.core.hub_scoping_migration import run_migration
334 from muse.core.bip39 import mnemonic_to_seed
335
336 identity_map = {
337 "musehub.ai": {
338 "type": "human",
339 "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")),
340 "fingerprint": "b" * 64,
341 }
342 }
343 seed = mnemonic_to_seed(FAKE_MNEMONIC)
344 hub_register = MagicMock()
345 result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
346 hub_register.assert_not_called()
347 assert result == []
348
349 def test_partial_failure_updates_successful_entries(self) -> None:
350 from muse.core.hub_scoping_migration import run_migration
351 from muse.core.bip39 import mnemonic_to_seed
352
353 identity_map = {
354 "ok.musehub.ai": {
355 "type": "human", "handle": "gabriel",
356 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
357 },
358 "bad.musehub.ai": {
359 "type": "human", "handle": "gabriel",
360 "hd_path": _pre_scoping_path(), "fingerprint": "b" * 64, "algorithm": "ed25519",
361 },
362 }
363 seed = mnemonic_to_seed(FAKE_MNEMONIC)
364
365 def _flaky_register(hub_key: str, new_fingerprint: str, new_hd_path: str, entry: Mapping[str, object]) -> bool:
366 if "bad" in hub_key:
367 raise RuntimeError("network error")
368 return True
369
370 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=_flaky_register, dry_run=False)
371
372 ok_result = next(r for r in results if r.hub_key == "ok.musehub.ai")
373 bad_result = next(r for r in results if r.hub_key == "bad.musehub.ai")
374 assert ok_result.hub_registered is True
375 assert bad_result.hub_registered is False
376 # Only the successful entry gets its hd_path/fingerprint updated locally --
377 # mutating a failed entry would leave identity.toml claiming a key the hub
378 # never actually received, permanently desyncing local from remote state.
379 assert identity_map["ok.musehub.ai"]["hd_path"] != _pre_scoping_path()
380 assert identity_map["bad.musehub.ai"]["hd_path"] == _pre_scoping_path()
381 assert identity_map["bad.musehub.ai"]["fingerprint"] == "b" * 64
382
383 def test_multi_hub_migrates_all_with_distinct_fingerprints(self) -> None:
384 from muse.core.hub_scoping_migration import run_migration
385 from muse.core.bip39 import mnemonic_to_seed
386
387 identity_map = {
388 "musehub.ai": {
389 "type": "human", "handle": "gabriel",
390 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
391 },
392 "staging.musehub.ai": {
393 "type": "human", "handle": "gabriel",
394 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
395 },
396 }
397 seed = mnemonic_to_seed(FAKE_MNEMONIC)
398 results = run_migration(
399 identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False
400 )
401 assert len(results) == 2
402 fps = {r.new_fingerprint for r in results}
403 assert len(fps) == 2, "each hub must get a distinct migrated fingerprint"
404
405 def test_skip_register_updates_local_state_without_calling_hub(self) -> None:
406 """skip_register=True (the --no-register case) is a deliberate choice, not a
407 failure -- local state should still be updated even though hub_register_fn is
408 never called."""
409 from muse.core.hub_scoping_migration import run_migration
410 from muse.core.bip39 import mnemonic_to_seed
411
412 identity_map = {
413 "musehub.ai": {
414 "type": "human", "handle": "gabriel",
415 "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519",
416 }
417 }
418 seed = mnemonic_to_seed(FAKE_MNEMONIC)
419 never_called = MagicMock()
420
421 results = run_migration(
422 identity_map=identity_map, seed=seed, hub_register_fn=never_called,
423 dry_run=False, skip_register=True,
424 )
425
426 never_called.assert_not_called()
427 assert results[0].hub_registered is False
428 assert identity_map["musehub.ai"]["hd_path"] == results[0].new_hd_path
429 assert identity_map["musehub.ai"]["fingerprint"] == results[0].new_fingerprint
430
431
432 # ============================================================================
433 # 6. CLI smoke
434 # ============================================================================
435
436
437 def _write_identity_toml(path: pathlib.Path, data: Mapping[str, Mapping[str, object]]) -> None:
438 lines = []
439 for section, fields in data.items():
440 lines.append(f'["{section}"]')
441 for k, v in fields.items():
442 lines.append(f'{k} = "{v}"')
443 lines.append("")
444 path.write_text("\n".join(lines), encoding="utf-8")
445 path.chmod(0o600)
446
447
448 class TestCliDryRun:
449 def test_cli_dry_run_exits_0_with_json(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
450 dot_muse = muse_dir(tmp_path)
451 dot_muse.mkdir()
452 identity_file = dot_muse / "identity.toml"
453 _write_identity_toml(identity_file, {
454 "musehub.ai": {
455 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
456 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
457 }
458 })
459
460 import muse.core.identity as id_module
461 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
462 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
463
464 import muse.core.keychain as kc_module
465 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
466
467 from tests.cli_test_helper import CliRunner
468 runner = CliRunner()
469 result = runner.invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"])
470
471 assert result.exit_code == 0, result.output
472 data = json.loads(result.output)
473 assert data["dry_run"] is True
474 assert data["entries_found"] == 1
475 assert data["entries_migrated"] == 0
476
477 def test_cli_no_pre_scoping_entries_exits_0(
478 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
479 ) -> None:
480 dot_muse = muse_dir(tmp_path)
481 dot_muse.mkdir()
482 identity_file = dot_muse / "identity.toml"
483 _write_identity_toml(identity_file, {
484 "musehub.ai": {
485 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
486 "fingerprint": "b" * 64,
487 "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")),
488 }
489 })
490
491 import muse.core.identity as id_module
492 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
493 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
494
495 import muse.core.keychain as kc_module
496 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
497
498 from tests.cli_test_helper import CliRunner
499 runner = CliRunner()
500 result = runner.invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"])
501
502 assert result.exit_code == 0, result.output
503 data = json.loads(result.output)
504 assert data["entries_found"] == 0
505
506 def test_cli_no_register_persists_to_identity_toml(
507 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
508 ) -> None:
509 dot_muse = muse_dir(tmp_path)
510 dot_muse.mkdir()
511 identity_file = dot_muse / "identity.toml"
512 _write_identity_toml(identity_file, {
513 "musehub.ai": {
514 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
515 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
516 }
517 })
518
519 import muse.core.identity as id_module
520 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
521 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
522
523 import muse.core.keychain as kc_module
524 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
525
526 from tests.cli_test_helper import CliRunner
527 runner = CliRunner()
528 result = runner.invoke(None, ["migrate", "hub-scoping", "--no-register", "--json"])
529 assert result.exit_code == 0, result.output
530
531 import tomllib
532 data = tomllib.loads(identity_file.read_text())
533 assert data["musehub.ai"]["hd_path"] != _pre_scoping_path()
534 assert "/" + str(hub_index("musehub.ai")) + "'" in data["musehub.ai"]["hd_path"]
535
536
537 # ============================================================================
538 # 7. Security: adversarial inputs
539 # ============================================================================
540
541
542 class TestSecurity:
543 def test_malformed_path_missing_hardened_marker_not_pre_scoping(self) -> None:
544 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
545 assert is_pre_hub_scoping_hd_path(f"m/{MUSE_PURPOSE}/{DOMAIN_IDENTITY}/0/0/0/0") is False
546
547 def test_path_with_too_few_segments_not_pre_scoping(self) -> None:
548 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
549 assert is_pre_hub_scoping_hd_path(f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/0'") is False
550
551 def test_path_with_extra_segments_not_pre_scoping(self) -> None:
552 """A path with 8+ segments is never mistaken for a pre-Phase-2 six-level path."""
553 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
554 assert is_pre_hub_scoping_hd_path(
555 f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/0'/0'/0'/0'/0'"
556 ) is False
557
558 def test_empty_string_not_pre_scoping(self) -> None:
559 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
560 assert is_pre_hub_scoping_hd_path("") is False
561
562 def test_whitespace_only_not_pre_scoping(self) -> None:
563 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
564 assert is_pre_hub_scoping_hd_path(" ") is False
565
566 def test_path_traversal_attempt_not_pre_scoping(self) -> None:
567 from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path
568 assert is_pre_hub_scoping_hd_path("m/1075233755'/../../../etc/passwd") is False
569
570 def test_new_path_hub_key_hashed_not_interpolated_raw(self) -> None:
571 """hub_key text never leaks verbatim into the derived path — only its hash does."""
572 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
573 malicious_hub = "musehub.ai/../../etc/passwd"
574 new = new_path_for_pre_hub_scoping(_pre_scoping_path(), malicious_hub)
575 assert "etc" not in new
576 assert "passwd" not in new
577 parts = new.split("/")[1:]
578 assert len(parts) == 7
579 assert all(p.endswith("'") and p[:-1].isdigit() for p in parts)
580
581
582 # ============================================================================
583 # 8. _make_hub_register_fn — real key-rotation wiring
584 #
585 # Regression coverage for TWO bugs found in sequence while planning
586 # musehub#221's real-world rollout:
587 #
588 # Bug 1 (first pass): the shim passed positional args that didn't match
589 # _post_challenge/_post_verify's actual (base_url, payload_dict) signatures,
590 # and never signed the challenge nonce at all. The resulting TypeError was
591 # silently swallowed, reporting "hub_registered=False" for every entry.
592 #
593 # Bug 2 (found running the *fixed* code for real against local musehub):
594 # POST /api/auth/verify is the fresh-registration endpoint. It correctly
595 # rejects a hub-scoping migration with HTTP 409 ("handle already taken"),
596 # because the handle is already registered under the pre-scoping key — this
597 # is a key *rotation* for an existing identity, not a new signup. The real
598 # endpoint is POST /api/auth/keys, MSign-authenticated with the OLD key
599 # (mirrors `muse auth rotate`). This also surfaced that _json_post_raw
600 # raises SystemExit (not a plain Exception) on HTTP failure, which the
601 # original `except Exception` wouldn't have caught either.
602 # ============================================================================
603
604
605 class TestMakeHubRegisterFn:
606 def _derive(self, seed: bytes, hd_path: str):
607 from muse.core.slip010 import derive_path, to_ed25519_private_key
608 dk = derive_path(seed, hd_path)
609 try:
610 return to_ed25519_private_key(dk)
611 finally:
612 dk.zero()
613
614 def _setup(self, hub: str = "musehub.ai"):
615 from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping
616 from muse.core.bip39 import mnemonic_to_seed
617 from muse.core.keypair import public_key_fingerprint
618
619 seed = mnemonic_to_seed(FAKE_MNEMONIC)
620 old_path = _pre_scoping_path()
621 new_path = new_path_for_pre_hub_scoping(old_path, hub)
622 old_key = self._derive(seed, old_path)
623 new_key = self._derive(seed, new_path)
624 old_fp = public_key_fingerprint(old_key.public_key())
625 new_fp = public_key_fingerprint(new_key.public_key())
626 entry = {"handle": "gabriel", "hd_path": old_path, "fingerprint": old_fp}
627 return seed, old_key, new_key, old_fp, new_fp, new_path, entry
628
629 def test_sends_correct_add_key_payload_and_msign_auth(self, monkeypatch: pytest.MonkeyPatch) -> None:
630 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
631 from muse.core.msign import verify_msign_header
632 from muse.core.keypair import public_key_to_b64url
633 from muse.core.types import DEFAULT_SIGN_ALGO
634
635 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
636
637 seen_challenge_payload = {}
638 add_key_call = {}
639
640 def _fake_challenge(base_url: str, payload: dict) -> dict:
641 seen_challenge_payload.update(payload)
642 return {"challenge_token": "ab" * 16, "is_new_key": True}
643
644 def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers: dict | None = None) -> dict:
645 add_key_call["base_url"] = base_url
646 add_key_call["path"] = path
647 add_key_call["payload"] = payload
648 add_key_call["extra_headers"] = extra_headers
649 return {}
650
651 monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _fake_challenge)
652 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw)
653 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)
654
655 register_fn = _make_hub_register_fn(seed, json_out=True)
656 ok = register_fn("musehub.ai", new_fp, new_path, entry)
657
658 assert ok is True
659 assert seen_challenge_payload["fingerprint"] == new_fp
660 assert seen_challenge_payload["algorithm"] == DEFAULT_SIGN_ALGO
661
662 assert add_key_call["path"] == "/api/auth/keys"
663 assert add_key_call["payload"]["public_key_b64"] == public_key_to_b64url(new_key.public_key())
664 assert add_key_call["payload"]["challenge_token"] == "ab" * 16
665
666 # The Authorization header must be a valid MSign signature by the OLD key
667 # (proof of account ownership) — not the new key.
668 from muse.core.types import split_pubkey
669
670 auth_header = add_key_call["extra_headers"]["Authorization"]
671 add_key_url = f"{add_key_call['base_url']}{add_key_call['path']}"
672 import json as _json
673 body_bytes = _json.dumps(add_key_call["payload"]).encode("utf-8")
674 _, old_pub_b64 = split_pubkey(public_key_to_b64url(old_key.public_key()))
675 verified, reason = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, old_pub_b64)
676 assert verified, reason
677
678 # It must NOT verify under the new key -- proves this isn't just
679 # coincidentally self-consistent.
680 _, new_pub_b64 = split_pubkey(public_key_to_b64url(new_key.public_key()))
681 verified_wrong, _ = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, new_pub_b64)
682 assert verified_wrong is False
683
684 def test_new_key_signature_in_payload_verifies_against_new_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
685 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
686 from muse.core.types import decode_sig
687
688 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
689 nonce_hex = "cd" * 16
690 add_key_payload = {}
691
692 monkeypatch.setattr(
693 "muse.cli.commands.auth._post_challenge",
694 lambda base_url, payload: {"challenge_token": nonce_hex, "is_new_key": True},
695 )
696
697 def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict:
698 add_key_payload.update(payload)
699 return {}
700
701 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw)
702 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)
703
704 register_fn = _make_hub_register_fn(seed, json_out=True)
705 ok = register_fn("musehub.ai", new_fp, new_path, entry)
706 assert ok is True
707
708 _, signature = decode_sig(add_key_payload["signature_b64"])
709 nonce_bytes = bytes.fromhex(nonce_hex)
710 # Raises InvalidSignature if this doesn't verify against the NEW key.
711 new_key.public_key().verify(signature, nonce_bytes)
712
713 def test_deregisters_old_key_after_successful_add(self, monkeypatch: pytest.MonkeyPatch) -> None:
714 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
715 from muse.cli.commands.auth import _compute_key_id
716 from muse.core.keypair import public_key_to_b64url
717 from muse.core.msign import verify_msign_header
718
719 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
720 delete_call = {}
721
722 monkeypatch.setattr(
723 "muse.cli.commands.auth._post_challenge",
724 lambda base_url, payload: {"challenge_token": "11" * 16, "is_new_key": True},
725 )
726 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
727
728 def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None:
729 delete_call["url"] = url
730 delete_call["auth_header"] = auth_header
731
732 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete)
733
734 register_fn = _make_hub_register_fn(seed, json_out=True)
735 ok = register_fn("musehub.ai", new_fp, new_path, entry)
736 assert ok is True
737
738 import urllib.parse
739 from muse.core.types import split_pubkey
740 old_pub_b64 = public_key_to_b64url(old_key.public_key())
741 expected_key_id = _compute_key_id(old_fp, old_pub_b64)
742 assert urllib.parse.quote(expected_key_id) in delete_call["url"]
743 assert "gabriel" in delete_call["url"]
744
745 _, old_pub_b64_bare = split_pubkey(old_pub_b64)
746 verified, reason = verify_msign_header(
747 delete_call["auth_header"], "DELETE", delete_call["url"], None, old_pub_b64_bare
748 )
749 assert verified, reason
750
751 def test_delete_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None:
752 """Old-key deregistration failing must not undo the fact that the new
753 key was already successfully registered -- mirrors `muse auth rotate`."""
754 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
755
756 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
757
758 monkeypatch.setattr(
759 "muse.cli.commands.auth._post_challenge",
760 lambda base_url, payload: {"challenge_token": "22" * 16, "is_new_key": True},
761 )
762 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
763
764 def _boom_delete(*a, **kw):
765 raise ConnectionError("hub unreachable for delete")
766
767 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _boom_delete)
768
769 register_fn = _make_hub_register_fn(seed, json_out=True)
770 ok = register_fn("musehub.ai", new_fp, new_path, entry)
771 assert ok is True
772
773 def test_missing_challenge_token_fails_closed(self, monkeypatch: pytest.MonkeyPatch) -> None:
774 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
775
776 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
777
778 monkeypatch.setattr(
779 "muse.cli.commands.auth._post_challenge",
780 lambda base_url, payload: {"challenge_token": "", "is_new_key": True},
781 )
782 add_key_called = []
783 monkeypatch.setattr(
784 "muse.cli.commands.auth._json_post_raw",
785 lambda *a, **kw: add_key_called.append(1) or {},
786 )
787
788 register_fn = _make_hub_register_fn(seed, json_out=True)
789 ok = register_fn("musehub.ai", new_fp, new_path, entry)
790
791 assert ok is False
792 assert add_key_called == []
793
794 def test_hub_http_failure_from_challenge_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None:
795 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
796
797 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
798
799 def _boom(base_url: str, payload: dict) -> dict:
800 raise ConnectionError("hub unreachable")
801
802 monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _boom)
803
804 register_fn = _make_hub_register_fn(seed, json_out=True)
805 ok = register_fn("musehub.ai", new_fp, new_path, entry)
806 assert ok is False
807
808 def test_systemexit_from_add_key_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None:
809 """_json_post_raw raises SystemExit (not a plain Exception) on a real
810 HTTP error -- e.g. the actual HTTP 409 hit in production testing when
811 this shim still called the wrong (fresh-registration) endpoint. A
812 multi-hub live run must not let one hub's HTTP error abort the whole
813 batch."""
814 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
815
816 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
817
818 monkeypatch.setattr(
819 "muse.cli.commands.auth._post_challenge",
820 lambda base_url, payload: {"challenge_token": "33" * 16, "is_new_key": True},
821 )
822
823 def _boom_post(*a, **kw):
824 raise SystemExit(1)
825
826 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _boom_post)
827
828 register_fn = _make_hub_register_fn(seed, json_out=True)
829 ok = register_fn("musehub.ai", new_fp, new_path, entry)
830 assert ok is False
831
832
833 # ============================================================================
834 # 9. CLI --hub filter
835 # ============================================================================
836
837
838 class TestCliHubFilter:
839 def test_hub_filter_restricts_to_named_hub(
840 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
841 ) -> None:
842 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
843 dot_muse = muse_dir(tmp_path)
844 dot_muse.mkdir()
845 identity_file = dot_muse / "identity.toml"
846 _write_identity_toml(identity_file, {
847 "musehub.ai": {
848 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
849 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
850 },
851 "staging.musehub.ai": {
852 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
853 "fingerprint": "b" * 64, "hd_path": _pre_scoping_path(),
854 },
855 })
856
857 import muse.core.identity as id_module
858 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
859 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
860
861 import muse.core.keychain as kc_module
862 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
863
864 from tests.cli_test_helper import CliRunner
865 runner = CliRunner()
866 result = runner.invoke(
867 None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://musehub.ai"]
868 )
869
870 assert result.exit_code == 0, result.output
871 data = json.loads(result.output)
872 assert data["entries_found"] == 1
873 assert data["results"][0]["hub_key"] == "musehub.ai"
874
875 def test_hub_filter_unknown_hub_errors(
876 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
877 ) -> None:
878 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
879 dot_muse = muse_dir(tmp_path)
880 dot_muse.mkdir()
881 identity_file = dot_muse / "identity.toml"
882 _write_identity_toml(identity_file, {
883 "musehub.ai": {
884 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
885 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
886 },
887 })
888
889 import muse.core.identity as id_module
890 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
891 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
892
893 import muse.core.keychain as kc_module
894 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
895
896 from tests.cli_test_helper import CliRunner
897 runner = CliRunner()
898 result = runner.invoke(
899 None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://nope.example.com"]
900 )
901
902 assert result.exit_code != 0
903
904
905 # ============================================================================
906 # 10. CLI live run — end-to-end through the real (fixed) registration shim
907 # ============================================================================
908
909
910 class TestCliLiveRunRegistersForReal:
911 def test_live_run_calls_real_challenge_and_verify_with_correct_payloads(
912 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
913 ) -> None:
914 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
915 dot_muse = muse_dir(tmp_path)
916 dot_muse.mkdir()
917 identity_file = dot_muse / "identity.toml"
918 _write_identity_toml(identity_file, {
919 "musehub.ai": {
920 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
921 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
922 },
923 })
924
925 import muse.core.identity as id_module
926 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
927 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
928
929 import muse.core.keychain as kc_module
930 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
931
932 challenge_calls = []
933 add_key_calls = []
934 delete_calls = []
935
936 def _fake_challenge(base_url: str, payload: dict) -> dict:
937 challenge_calls.append((base_url, payload))
938 return {"challenge_token": "ef" * 16, "is_new_key": True}
939
940 def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict:
941 add_key_calls.append((base_url, path, payload, extra_headers))
942 return {}
943
944 def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None:
945 delete_calls.append((url, auth_header))
946
947 monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _fake_challenge)
948 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw)
949 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete)
950
951 from tests.cli_test_helper import CliRunner
952 runner = CliRunner()
953 result = runner.invoke(None, ["migrate", "hub-scoping", "--json"])
954
955 assert result.exit_code == 0, result.output
956 data = json.loads(result.output)
957 assert data["entries_migrated"] == 1
958 assert data["results"][0]["hub_registered"] is True
959
960 assert len(challenge_calls) == 1
961 assert len(add_key_calls) == 1
962 challenge_payload = challenge_calls[0][1]
963 _, add_key_path, add_key_payload, add_key_headers = add_key_calls[0]
964 assert challenge_payload["fingerprint"] == data["results"][0]["new_fingerprint"]
965 assert add_key_path == "/api/auth/keys"
966 assert add_key_payload["challenge_token"] == "ef" * 16
967 assert "public_key_b64" in add_key_payload
968 assert "signature_b64" in add_key_payload
969 assert add_key_headers["Authorization"].startswith("MSign ")
970
971 # Old key deregistration was attempted, signed by the old key too.
972 assert len(delete_calls) == 1
973 assert "gabriel" in delete_calls[0][0]
974 assert delete_calls[0][1].startswith("MSign ")
975
976 import tomllib
977 toml_data = tomllib.loads(identity_file.read_text())
978 assert toml_data["musehub.ai"]["fingerprint"] == data["results"][0]["new_fingerprint"]
File History 1 commit
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ 8 hours ago