gabriel / muse public
test_migrate_hub_scoping.py python
1,065 lines 46.4 KB
Raw
sha256:bfd014ea21257cca982f8e389be81f60616acc78fb3d84f767601fea6d35a7f5 fix: look up old key_id via hub instead of recomputing it (… Sonnet 5 1 day 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(
654 "muse.cli.commands.auth._hub_get",
655 lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "9" * 64, "fingerprint": old_fp}]},
656 )
657 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)
658
659 register_fn = _make_hub_register_fn(seed, json_out=True)
660 ok = register_fn("musehub.ai", new_fp, new_path, entry)
661
662 assert ok is True
663 assert seen_challenge_payload["fingerprint"] == new_fp
664 assert seen_challenge_payload["algorithm"] == DEFAULT_SIGN_ALGO
665
666 assert add_key_call["path"] == "/api/auth/keys"
667 assert add_key_call["payload"]["public_key_b64"] == public_key_to_b64url(new_key.public_key())
668 assert add_key_call["payload"]["challenge_token"] == "ab" * 16
669
670 # The Authorization header must be a valid MSign signature by the OLD key
671 # (proof of account ownership) — not the new key.
672 from muse.core.types import split_pubkey
673
674 auth_header = add_key_call["extra_headers"]["Authorization"]
675 add_key_url = f"{add_key_call['base_url']}{add_key_call['path']}"
676 import json as _json
677 body_bytes = _json.dumps(add_key_call["payload"]).encode("utf-8")
678 _, old_pub_b64 = split_pubkey(public_key_to_b64url(old_key.public_key()))
679 verified, reason = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, old_pub_b64)
680 assert verified, reason
681
682 # It must NOT verify under the new key -- proves this isn't just
683 # coincidentally self-consistent.
684 _, new_pub_b64 = split_pubkey(public_key_to_b64url(new_key.public_key()))
685 verified_wrong, _ = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, new_pub_b64)
686 assert verified_wrong is False
687
688 def test_new_key_signature_in_payload_verifies_against_new_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
689 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
690 from muse.core.types import decode_sig
691
692 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
693 nonce_hex = "cd" * 16
694 add_key_payload = {}
695
696 monkeypatch.setattr(
697 "muse.cli.commands.auth._post_challenge",
698 lambda base_url, payload: {"challenge_token": nonce_hex, "is_new_key": True},
699 )
700
701 def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict:
702 add_key_payload.update(payload)
703 return {}
704
705 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw)
706 monkeypatch.setattr(
707 "muse.cli.commands.auth._hub_get",
708 lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "9" * 64, "fingerprint": old_fp}]},
709 )
710 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None)
711
712 register_fn = _make_hub_register_fn(seed, json_out=True)
713 ok = register_fn("musehub.ai", new_fp, new_path, entry)
714 assert ok is True
715
716 _, signature = decode_sig(add_key_payload["signature_b64"])
717 nonce_bytes = bytes.fromhex(nonce_hex)
718 # Raises InvalidSignature if this doesn't verify against the NEW key.
719 new_key.public_key().verify(signature, nonce_bytes)
720
721 def test_deregisters_old_key_after_successful_add(self, monkeypatch: pytest.MonkeyPatch) -> None:
722 """The old key's key_id is looked up via GET /api/auth/keys/{handle}, matched
723 by fingerprint -- NOT recomputed locally. Recomputing depends on reproducing
724 the exact public_key_b64 encoding the account's original registration used,
725 which silently 404'd in production for accounts registered under an older
726 encoding convention (found running this for real against musehub.ai)."""
727 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
728 from muse.core.keypair import public_key_to_b64url
729 from muse.core.msign import verify_msign_header
730 from muse.core.types import split_pubkey
731
732 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
733 real_old_key_id = "sha256:" + "7" * 64 # deliberately NOT what _compute_key_id would derive
734 list_call = {}
735 delete_call = {}
736
737 monkeypatch.setattr(
738 "muse.cli.commands.auth._post_challenge",
739 lambda base_url, payload: {"challenge_token": "11" * 16, "is_new_key": True},
740 )
741 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
742
743 def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict:
744 list_call["url"] = url
745 list_call["auth_header"] = auth_header
746 return {"keys": [
747 {"key_id": real_old_key_id, "fingerprint": old_fp},
748 {"key_id": "sha256:" + "8" * 64, "fingerprint": new_fp},
749 ]}
750
751 def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None:
752 delete_call["url"] = url
753 delete_call["auth_header"] = auth_header
754
755 monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get)
756 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete)
757
758 register_fn = _make_hub_register_fn(seed, json_out=True)
759 ok = register_fn("musehub.ai", new_fp, new_path, entry)
760 assert ok is True
761
762 import urllib.parse
763 assert "gabriel" in list_call["url"]
764 assert urllib.parse.quote(real_old_key_id) in delete_call["url"]
765 assert "gabriel" in delete_call["url"]
766
767 _, old_pub_b64_bare = split_pubkey(public_key_to_b64url(old_key.public_key()))
768 verified, reason = verify_msign_header(
769 list_call["auth_header"], "GET", list_call["url"], None, old_pub_b64_bare
770 )
771 assert verified, reason
772 verified, reason = verify_msign_header(
773 delete_call["auth_header"], "DELETE", delete_call["url"], None, old_pub_b64_bare
774 )
775 assert verified, reason
776
777 def test_delete_skipped_when_old_key_not_found_in_hub_list(self, monkeypatch: pytest.MonkeyPatch) -> None:
778 """If the hub's key list doesn't contain the old fingerprint at all, deletion
779 is skipped (nothing to delete) rather than guessing a key_id -- still non-fatal."""
780 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
781
782 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
783 delete_called = []
784
785 monkeypatch.setattr(
786 "muse.cli.commands.auth._post_challenge",
787 lambda base_url, payload: {"challenge_token": "33" * 16, "is_new_key": True},
788 )
789 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
790 monkeypatch.setattr(
791 "muse.cli.commands.auth._hub_get",
792 lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "8" * 64, "fingerprint": new_fp}]},
793 )
794 monkeypatch.setattr(
795 "muse.cli.commands.auth._hub_delete",
796 lambda *a, **kw: delete_called.append(1),
797 )
798
799 register_fn = _make_hub_register_fn(seed, json_out=True)
800 ok = register_fn("musehub.ai", new_fp, new_path, entry)
801 assert ok is True
802 assert delete_called == []
803
804 def test_delete_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None:
805 """Old-key deregistration failing must not undo the fact that the new
806 key was already successfully registered -- mirrors `muse auth rotate`."""
807 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
808
809 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
810
811 monkeypatch.setattr(
812 "muse.cli.commands.auth._post_challenge",
813 lambda base_url, payload: {"challenge_token": "22" * 16, "is_new_key": True},
814 )
815 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
816 monkeypatch.setattr(
817 "muse.cli.commands.auth._hub_get",
818 lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": old_fp}]},
819 )
820
821 def _boom_delete(*a, **kw):
822 raise ConnectionError("hub unreachable for delete")
823
824 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _boom_delete)
825
826 register_fn = _make_hub_register_fn(seed, json_out=True)
827 ok = register_fn("musehub.ai", new_fp, new_path, entry)
828 assert ok is True
829
830 def test_lookup_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None:
831 """If GET /api/auth/keys/{handle} itself fails, deregistration is skipped
832 entirely (non-fatal) -- the new key is already registered regardless."""
833 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
834
835 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
836
837 monkeypatch.setattr(
838 "muse.cli.commands.auth._post_challenge",
839 lambda base_url, payload: {"challenge_token": "44" * 16, "is_new_key": True},
840 )
841 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {})
842
843 def _boom_get(*a, **kw):
844 raise ConnectionError("hub unreachable for key list")
845
846 monkeypatch.setattr("muse.cli.commands.auth._hub_get", _boom_get)
847
848 register_fn = _make_hub_register_fn(seed, json_out=True)
849 ok = register_fn("musehub.ai", new_fp, new_path, entry)
850 assert ok is True
851
852 def test_missing_challenge_token_fails_closed(self, monkeypatch: pytest.MonkeyPatch) -> None:
853 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
854
855 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
856
857 monkeypatch.setattr(
858 "muse.cli.commands.auth._post_challenge",
859 lambda base_url, payload: {"challenge_token": "", "is_new_key": True},
860 )
861 add_key_called = []
862 monkeypatch.setattr(
863 "muse.cli.commands.auth._json_post_raw",
864 lambda *a, **kw: add_key_called.append(1) or {},
865 )
866
867 register_fn = _make_hub_register_fn(seed, json_out=True)
868 ok = register_fn("musehub.ai", new_fp, new_path, entry)
869
870 assert ok is False
871 assert add_key_called == []
872
873 def test_hub_http_failure_from_challenge_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None:
874 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
875
876 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
877
878 def _boom(base_url: str, payload: dict) -> dict:
879 raise ConnectionError("hub unreachable")
880
881 monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _boom)
882
883 register_fn = _make_hub_register_fn(seed, json_out=True)
884 ok = register_fn("musehub.ai", new_fp, new_path, entry)
885 assert ok is False
886
887 def test_systemexit_from_add_key_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None:
888 """_json_post_raw raises SystemExit (not a plain Exception) on a real
889 HTTP error -- e.g. the actual HTTP 409 hit in production testing when
890 this shim still called the wrong (fresh-registration) endpoint. A
891 multi-hub live run must not let one hub's HTTP error abort the whole
892 batch."""
893 from muse.cli.commands.migrate_cmd import _make_hub_register_fn
894
895 seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup()
896
897 monkeypatch.setattr(
898 "muse.cli.commands.auth._post_challenge",
899 lambda base_url, payload: {"challenge_token": "33" * 16, "is_new_key": True},
900 )
901
902 def _boom_post(*a, **kw):
903 raise SystemExit(1)
904
905 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _boom_post)
906
907 register_fn = _make_hub_register_fn(seed, json_out=True)
908 ok = register_fn("musehub.ai", new_fp, new_path, entry)
909 assert ok is False
910
911
912 # ============================================================================
913 # 9. CLI --hub filter
914 # ============================================================================
915
916
917 class TestCliHubFilter:
918 def test_hub_filter_restricts_to_named_hub(
919 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
920 ) -> None:
921 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
922 dot_muse = muse_dir(tmp_path)
923 dot_muse.mkdir()
924 identity_file = dot_muse / "identity.toml"
925 _write_identity_toml(identity_file, {
926 "musehub.ai": {
927 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
928 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
929 },
930 "staging.musehub.ai": {
931 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
932 "fingerprint": "b" * 64, "hd_path": _pre_scoping_path(),
933 },
934 })
935
936 import muse.core.identity as id_module
937 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
938 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
939
940 import muse.core.keychain as kc_module
941 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
942
943 from tests.cli_test_helper import CliRunner
944 runner = CliRunner()
945 result = runner.invoke(
946 None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://musehub.ai"]
947 )
948
949 assert result.exit_code == 0, result.output
950 data = json.loads(result.output)
951 assert data["entries_found"] == 1
952 assert data["results"][0]["hub_key"] == "musehub.ai"
953
954 def test_hub_filter_unknown_hub_errors(
955 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
956 ) -> None:
957 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
958 dot_muse = muse_dir(tmp_path)
959 dot_muse.mkdir()
960 identity_file = dot_muse / "identity.toml"
961 _write_identity_toml(identity_file, {
962 "musehub.ai": {
963 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
964 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
965 },
966 })
967
968 import muse.core.identity as id_module
969 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
970 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
971
972 import muse.core.keychain as kc_module
973 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
974
975 from tests.cli_test_helper import CliRunner
976 runner = CliRunner()
977 result = runner.invoke(
978 None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://nope.example.com"]
979 )
980
981 assert result.exit_code != 0
982
983
984 # ============================================================================
985 # 10. CLI live run — end-to-end through the real (fixed) registration shim
986 # ============================================================================
987
988
989 class TestCliLiveRunRegistersForReal:
990 def test_live_run_calls_real_challenge_and_verify_with_correct_payloads(
991 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
992 ) -> None:
993 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
994 dot_muse = muse_dir(tmp_path)
995 dot_muse.mkdir()
996 identity_file = dot_muse / "identity.toml"
997 _write_identity_toml(identity_file, {
998 "musehub.ai": {
999 "type": "human", "handle": "gabriel", "algorithm": "ed25519",
1000 "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(),
1001 },
1002 })
1003
1004 import muse.core.identity as id_module
1005 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
1006 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
1007
1008 import muse.core.keychain as kc_module
1009 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
1010
1011 challenge_calls = []
1012 add_key_calls = []
1013 list_calls = []
1014 delete_calls = []
1015
1016 def _fake_challenge(base_url: str, payload: dict) -> dict:
1017 challenge_calls.append((base_url, payload))
1018 return {"challenge_token": "ef" * 16, "is_new_key": True}
1019
1020 def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict:
1021 add_key_calls.append((base_url, path, payload, extra_headers))
1022 return {}
1023
1024 def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict:
1025 list_calls.append((url, auth_header))
1026 # The fixture identity entry above has fingerprint "a" * 64 -- the old
1027 # (pre-migration) fingerprint the migration will look up by.
1028 return {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": "a" * 64}]}
1029
1030 def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None:
1031 delete_calls.append((url, auth_header))
1032
1033 monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _fake_challenge)
1034 monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw)
1035 monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get)
1036 monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete)
1037
1038 from tests.cli_test_helper import CliRunner
1039 runner = CliRunner()
1040 result = runner.invoke(None, ["migrate", "hub-scoping", "--json"])
1041
1042 assert result.exit_code == 0, result.output
1043 data = json.loads(result.output)
1044 assert data["entries_migrated"] == 1
1045 assert data["results"][0]["hub_registered"] is True
1046
1047 assert len(challenge_calls) == 1
1048 assert len(add_key_calls) == 1
1049 challenge_payload = challenge_calls[0][1]
1050 _, add_key_path, add_key_payload, add_key_headers = add_key_calls[0]
1051 assert challenge_payload["fingerprint"] == data["results"][0]["new_fingerprint"]
1052 assert add_key_path == "/api/auth/keys"
1053 assert add_key_payload["challenge_token"] == "ef" * 16
1054 assert "public_key_b64" in add_key_payload
1055 assert "signature_b64" in add_key_payload
1056 assert add_key_headers["Authorization"].startswith("MSign ")
1057
1058 # Old key deregistration was attempted, signed by the old key too.
1059 assert len(delete_calls) == 1
1060 assert "gabriel" in delete_calls[0][0]
1061 assert delete_calls[0][1].startswith("MSign ")
1062
1063 import tomllib
1064 toml_data = tomllib.loads(identity_file.read_text())
1065 assert toml_data["musehub.ai"]["fingerprint"] == data["results"][0]["new_fingerprint"]
File History 1 commit
sha256:bfd014ea21257cca982f8e389be81f60616acc78fb3d84f767601fea6d35a7f5 fix: look up old key_id via hub instead of recomputing it (… Sonnet 5 1 day ago