"""TDD tests for ``muse migrate hub-scoping`` (musehub#221). Covers: 1. Core library (muse.core.hub_scoping_migration) - Pre-Phase-2 path detection (identity domain only) - Old -> new path mapping, hub-scoped - Key re-derivation produces a different (correct) fingerprint per hub - Identity file scanning 2. Dry-run plan (no writes, no hub calls) 3. Live run (hub called, identity map mutated) 4. CLI smoke (muse migrate hub-scoping --dry-run / --no-register) 5. Security adversarial inputs and boundary conditions Background ---------- Prior to Phase 2, the identity key at rotation index 0 was bit-for-bit identical regardless of which hub it was registered with. Phase 2 inserts a hardened ``hub'`` level between ``role'`` and ``index'``. Users with keys derived before Phase 2 must re-derive at the new, hub-scoped path and re-register with the affected hub. """ from __future__ import annotations import json import pathlib from collections.abc import Mapping from unittest.mock import MagicMock import pytest from muse.core.hdkeys import ( DOMAIN_IDENTITY, DOMAIN_CODE, ENTITY_AGENT, ROLE_ATTEST, hub_index, muse_path, ) from muse.core.paths import muse_dir from muse.core.slip010 import MUSE_PURPOSE FAKE_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon " "abandon abandon abandon abandon abandon about" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _pre_scoping_path(entity_type: int = 0, entity_id: int = 0, role: int = 0, index: int = 0) -> str: """Build a pre-Phase-2 six-level identity path (no hub segment).""" return muse_path(DOMAIN_IDENTITY, entity_type, entity_id, role, index) # ============================================================================ # 1. Core: pre-Phase-2 path detection # ============================================================================ class TestIsPreHubScopingHdPath: def test_six_level_identity_path_is_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path(_pre_scoping_path()) is True def test_seven_level_hub_scoped_path_is_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path hub = hub_index("musehub.ai") scoped = muse_path(DOMAIN_IDENTITY, hub=hub) assert is_pre_hub_scoping_hd_path(scoped) is False def test_non_identity_domain_six_level_path_is_not_flagged(self) -> None: """Code/music/etc. domains never had hub scoping — not part of this migration.""" from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path(muse_path(DOMAIN_CODE)) is False def test_empty_string_is_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path("") is False def test_non_muse_purpose_is_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path("m/44'/0'/0'/0'/0'/0'") is False def test_agent_pre_scoping_path_is_flagged(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path( _pre_scoping_path(entity_type=ENTITY_AGENT, entity_id=3) ) is True # ============================================================================ # 2. Core: old -> new path mapping # ============================================================================ class TestNewPathForPreHubScoping: def test_inserts_hub_segment_before_index(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping old = _pre_scoping_path() new = new_path_for_pre_hub_scoping(old, "musehub.ai") expected_hub = hub_index("musehub.ai") assert new == muse_path(DOMAIN_IDENTITY, hub=expected_hub) def test_different_hubs_produce_different_paths(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping old = _pre_scoping_path() a = new_path_for_pre_hub_scoping(old, "musehub.ai") b = new_path_for_pre_hub_scoping(old, "staging.musehub.ai") assert a != b def test_preserves_entity_type_and_id(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping old = _pre_scoping_path(entity_type=ENTITY_AGENT, entity_id=5) new = new_path_for_pre_hub_scoping(old, "musehub.ai") parts = new.split("/") assert parts[3] == f"{ENTITY_AGENT}'" assert parts[4] == "5'" def test_preserves_role_and_index(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping old = _pre_scoping_path(role=ROLE_ATTEST, index=2) new = new_path_for_pre_hub_scoping(old, "musehub.ai") parts = new.split("/") assert parts[5] == f"{ROLE_ATTEST}'" assert parts[-1] == "2'" def test_non_identity_domain_raises(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping with pytest.raises(ValueError, match="Not an identity-domain path"): new_path_for_pre_hub_scoping(muse_path(DOMAIN_CODE), "musehub.ai") def test_already_hub_scoped_path_raises(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping scoped = muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")) with pytest.raises(ValueError): new_path_for_pre_hub_scoping(scoped, "musehub.ai") def test_output_has_seven_hardened_segments(self) -> None: from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping new = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") assert new.startswith(f"m/{MUSE_PURPOSE}'") parts = new.split("/")[1:] assert len(parts) == 7 assert all(p.endswith("'") for p in parts) # ============================================================================ # 3. Core: key re-derivation # ============================================================================ class TestDeriveFingerprintAtHubScopedPath: def test_old_and_new_fingerprints_differ(self) -> None: from muse.core.hub_scoping_migration import ( derive_fingerprint_at_hub_scoped_path, new_path_for_pre_hub_scoping, ) from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) old_fp = derive_fingerprint_at_path(seed, _pre_scoping_path()) new_path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") new_fp = derive_fingerprint_at_hub_scoped_path(seed, new_path) assert old_fp != new_fp def test_different_hubs_produce_different_fingerprints(self) -> None: from muse.core.hub_scoping_migration import ( derive_fingerprint_at_hub_scoped_path, new_path_for_pre_hub_scoping, ) from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) path_a = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") path_b = new_path_for_pre_hub_scoping(_pre_scoping_path(), "staging.musehub.ai") fp_a = derive_fingerprint_at_hub_scoped_path(seed, path_a) fp_b = derive_fingerprint_at_hub_scoped_path(seed, path_b) assert fp_a != fp_b def test_deterministic(self) -> None: from muse.core.hub_scoping_migration import ( derive_fingerprint_at_hub_scoped_path, new_path_for_pre_hub_scoping, ) from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") fp1 = derive_fingerprint_at_hub_scoped_path(seed, path) fp2 = derive_fingerprint_at_hub_scoped_path(seed, path) assert fp1 == fp2 def test_matches_direct_derive_identity_key(self) -> None: from muse.core.hub_scoping_migration import derive_fingerprint_at_hub_scoped_path from muse.core.bip39 import mnemonic_to_seed from muse.core.keypair import derive_hd_public_info seed = mnemonic_to_seed(FAKE_MNEMONIC) hub = hub_index("musehub.ai") _, expected_fp = derive_hd_public_info(seed, hub=hub) actual_fp = derive_fingerprint_at_hub_scoped_path(seed, muse_path(DOMAIN_IDENTITY, hub=hub)) assert actual_fp == expected_fp def test_fingerprint_is_sha256_prefixed(self) -> None: from muse.core.hub_scoping_migration import ( derive_fingerprint_at_hub_scoped_path, new_path_for_pre_hub_scoping, ) from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") fp = derive_fingerprint_at_hub_scoped_path(seed, path) assert fp.startswith("sha256:") assert len(fp) == 71 def test_rejects_six_level_path(self) -> None: from muse.core.hub_scoping_migration import derive_fingerprint_at_hub_scoped_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) with pytest.raises(ValueError, match="Cannot parse"): derive_fingerprint_at_hub_scoped_path(seed, _pre_scoping_path()) # ============================================================================ # 4. Core: scanning identity map # ============================================================================ class TestScanForPreHubScoping: def test_finds_pre_scoping_entry(self) -> None: from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping identity_map = { "musehub.ai": {"type": "human", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64}, } assert "musehub.ai" in scan_for_pre_hub_scoping(identity_map) def test_ignores_already_scoped_entry(self) -> None: from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping identity_map = { "musehub.ai": { "type": "human", "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")), "fingerprint": "b" * 64, }, } assert scan_for_pre_hub_scoping(identity_map) == [] def test_ignores_non_identity_domain_entry(self) -> None: from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping identity_map = {"musehub.ai": {"hd_path": muse_path(DOMAIN_CODE), "fingerprint": "c" * 64}} assert scan_for_pre_hub_scoping(identity_map) == [] def test_finds_multiple_hubs(self) -> None: from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping identity_map = { "musehub.ai": {"hd_path": _pre_scoping_path(), "fingerprint": "a" * 64}, "staging.musehub.ai": {"hd_path": _pre_scoping_path(), "fingerprint": "b" * 64}, } assert set(scan_for_pre_hub_scoping(identity_map)) == {"musehub.ai", "staging.musehub.ai"} def test_empty_map_returns_empty(self) -> None: from muse.core.hub_scoping_migration import scan_for_pre_hub_scoping assert scan_for_pre_hub_scoping({}) == [] # ============================================================================ # 5. Dry-run and live run # ============================================================================ class TestDryRun: def test_dry_run_returns_plans_without_registering(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock() result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=True) hub_register.assert_not_called() assert len(result) == 1 assert result[0].hub_registered is False assert result[0].new_fingerprint != result[0].old_fingerprint def test_dry_run_does_not_mutate_identity_map(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": {"type": "human", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64}, } seed = mnemonic_to_seed(FAKE_MNEMONIC) run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True) assert identity_map["musehub.ai"]["hd_path"] == _pre_scoping_path() assert identity_map["musehub.ai"]["fingerprint"] == "a" * 64 class TestLiveMigration: def test_live_run_calls_hub_register_and_updates_map(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(return_value=True) results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) hub_register.assert_called_once() assert results[0].hub_registered is True assert identity_map["musehub.ai"]["hd_path"] == results[0].new_hd_path assert identity_map["musehub.ai"]["fingerprint"] == results[0].new_fingerprint def test_live_run_skips_already_scoped_entries(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": { "type": "human", "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")), "fingerprint": "b" * 64, } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock() result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) hub_register.assert_not_called() assert result == [] def test_partial_failure_updates_successful_entries(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "ok.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", }, "bad.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "b" * 64, "algorithm": "ed25519", }, } seed = mnemonic_to_seed(FAKE_MNEMONIC) def _flaky_register(hub_key: str, new_fingerprint: str, new_hd_path: str, entry: Mapping[str, object]) -> bool: if "bad" in hub_key: raise RuntimeError("network error") return True results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=_flaky_register, dry_run=False) ok_result = next(r for r in results if r.hub_key == "ok.musehub.ai") bad_result = next(r for r in results if r.hub_key == "bad.musehub.ai") assert ok_result.hub_registered is True assert bad_result.hub_registered is False # Only the successful entry gets its hd_path/fingerprint updated locally -- # mutating a failed entry would leave identity.toml claiming a key the hub # never actually received, permanently desyncing local from remote state. assert identity_map["ok.musehub.ai"]["hd_path"] != _pre_scoping_path() assert identity_map["bad.musehub.ai"]["hd_path"] == _pre_scoping_path() assert identity_map["bad.musehub.ai"]["fingerprint"] == "b" * 64 def test_multi_hub_migrates_all_with_distinct_fingerprints(self) -> None: from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", }, "staging.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", }, } seed = mnemonic_to_seed(FAKE_MNEMONIC) results = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False ) assert len(results) == 2 fps = {r.new_fingerprint for r in results} assert len(fps) == 2, "each hub must get a distinct migrated fingerprint" def test_skip_register_updates_local_state_without_calling_hub(self) -> None: """skip_register=True (the --no-register case) is a deliberate choice, not a failure -- local state should still be updated even though hub_register_fn is never called.""" from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) never_called = MagicMock() results = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=never_called, dry_run=False, skip_register=True, ) never_called.assert_not_called() assert results[0].hub_registered is False assert identity_map["musehub.ai"]["hd_path"] == results[0].new_hd_path assert identity_map["musehub.ai"]["fingerprint"] == results[0].new_fingerprint # ============================================================================ # 6. CLI smoke # ============================================================================ def _write_identity_toml(path: pathlib.Path, data: Mapping[str, Mapping[str, object]]) -> None: lines = [] for section, fields in data.items(): lines.append(f'["{section}"]') for k, v in fields.items(): lines.append(f'{k} = "{v}"') lines.append("") path.write_text("\n".join(lines), encoding="utf-8") path.chmod(0o600) class TestCliDryRun: def test_cli_dry_run_exits_0_with_json(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), } }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"]) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["dry_run"] is True assert data["entries_found"] == 1 assert data["entries_migrated"] == 0 def test_cli_no_pre_scoping_entries_exits_0( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "b" * 64, "hd_path": muse_path(DOMAIN_IDENTITY, hub=hub_index("musehub.ai")), } }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"]) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["entries_found"] == 0 def test_cli_no_register_persists_to_identity_toml( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), } }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "hub-scoping", "--no-register", "--json"]) assert result.exit_code == 0, result.output import tomllib data = tomllib.loads(identity_file.read_text()) assert data["musehub.ai"]["hd_path"] != _pre_scoping_path() assert "/" + str(hub_index("musehub.ai")) + "'" in data["musehub.ai"]["hd_path"] # ============================================================================ # 7. Security: adversarial inputs # ============================================================================ class TestSecurity: def test_malformed_path_missing_hardened_marker_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path(f"m/{MUSE_PURPOSE}/{DOMAIN_IDENTITY}/0/0/0/0") is False def test_path_with_too_few_segments_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path(f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/0'") is False def test_path_with_extra_segments_not_pre_scoping(self) -> None: """A path with 8+ segments is never mistaken for a pre-Phase-2 six-level path.""" from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path( f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/0'/0'/0'/0'/0'" ) is False def test_empty_string_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path("") is False def test_whitespace_only_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path(" ") is False def test_path_traversal_attempt_not_pre_scoping(self) -> None: from muse.core.hub_scoping_migration import is_pre_hub_scoping_hd_path assert is_pre_hub_scoping_hd_path("m/1075233755'/../../../etc/passwd") is False def test_new_path_hub_key_hashed_not_interpolated_raw(self) -> None: """hub_key text never leaks verbatim into the derived path — only its hash does.""" from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping malicious_hub = "musehub.ai/../../etc/passwd" new = new_path_for_pre_hub_scoping(_pre_scoping_path(), malicious_hub) assert "etc" not in new assert "passwd" not in new parts = new.split("/")[1:] assert len(parts) == 7 assert all(p.endswith("'") and p[:-1].isdigit() for p in parts) # ============================================================================ # 8. _make_hub_register_fn — real key-rotation wiring # # Regression coverage for TWO bugs found in sequence while planning # musehub#221's real-world rollout: # # Bug 1 (first pass): the shim passed positional args that didn't match # _post_challenge/_post_verify's actual (base_url, payload_dict) signatures, # and never signed the challenge nonce at all. The resulting TypeError was # silently swallowed, reporting "hub_registered=False" for every entry. # # Bug 2 (found running the *fixed* code for real against local musehub): # POST /api/auth/verify is the fresh-registration endpoint. It correctly # rejects a hub-scoping migration with HTTP 409 ("handle already taken"), # because the handle is already registered under the pre-scoping key — this # is a key *rotation* for an existing identity, not a new signup. The real # endpoint is POST /api/auth/keys, MSign-authenticated with the OLD key # (mirrors `muse auth rotate`). This also surfaced that _json_post_raw # raises SystemExit (not a plain Exception) on HTTP failure, which the # original `except Exception` wouldn't have caught either. # ============================================================================ class TestMakeHubRegisterFn: def _derive(self, seed: bytes, hd_path: str): from muse.core.slip010 import derive_path, to_ed25519_private_key dk = derive_path(seed, hd_path) try: return to_ed25519_private_key(dk) finally: dk.zero() def _setup(self, hub: str = "musehub.ai"): from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping from muse.core.bip39 import mnemonic_to_seed from muse.core.keypair import public_key_fingerprint seed = mnemonic_to_seed(FAKE_MNEMONIC) old_path = _pre_scoping_path() new_path = new_path_for_pre_hub_scoping(old_path, hub) old_key = self._derive(seed, old_path) new_key = self._derive(seed, new_path) old_fp = public_key_fingerprint(old_key.public_key()) new_fp = public_key_fingerprint(new_key.public_key()) entry = {"handle": "gabriel", "hd_path": old_path, "fingerprint": old_fp} return seed, old_key, new_key, old_fp, new_fp, new_path, entry def test_sends_correct_add_key_payload_and_msign_auth(self, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.commands.migrate_cmd import _make_hub_register_fn from muse.core.msign import verify_msign_header from muse.core.keypair import public_key_to_b64url from muse.core.types import DEFAULT_SIGN_ALGO seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() seen_challenge_payload = {} add_key_call = {} def _fake_challenge(base_url: str, payload: dict) -> dict: seen_challenge_payload.update(payload) return {"challenge_token": "ab" * 16, "is_new_key": True} def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers: dict | None = None) -> dict: add_key_call["base_url"] = base_url add_key_call["path"] = path add_key_call["payload"] = payload add_key_call["extra_headers"] = extra_headers return {} monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _fake_challenge) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw) monkeypatch.setattr( "muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "9" * 64, "fingerprint": old_fp}]}, ) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True assert seen_challenge_payload["fingerprint"] == new_fp assert seen_challenge_payload["algorithm"] == DEFAULT_SIGN_ALGO assert add_key_call["path"] == "/api/auth/keys" assert add_key_call["payload"]["public_key_b64"] == public_key_to_b64url(new_key.public_key()) assert add_key_call["payload"]["challenge_token"] == "ab" * 16 # The Authorization header must be a valid MSign signature by the OLD key # (proof of account ownership) — not the new key. from muse.core.types import split_pubkey auth_header = add_key_call["extra_headers"]["Authorization"] add_key_url = f"{add_key_call['base_url']}{add_key_call['path']}" import json as _json body_bytes = _json.dumps(add_key_call["payload"]).encode("utf-8") _, old_pub_b64 = split_pubkey(public_key_to_b64url(old_key.public_key())) verified, reason = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, old_pub_b64) assert verified, reason # It must NOT verify under the new key -- proves this isn't just # coincidentally self-consistent. _, new_pub_b64 = split_pubkey(public_key_to_b64url(new_key.public_key())) verified_wrong, _ = verify_msign_header(auth_header, "POST", add_key_url, body_bytes, new_pub_b64) assert verified_wrong is False def test_new_key_signature_in_payload_verifies_against_new_key(self, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.commands.migrate_cmd import _make_hub_register_fn from muse.core.types import decode_sig seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() nonce_hex = "cd" * 16 add_key_payload = {} monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": nonce_hex, "is_new_key": True}, ) def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict: add_key_payload.update(payload) return {} monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw) monkeypatch.setattr( "muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "9" * 64, "fingerprint": old_fp}]}, ) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True _, signature = decode_sig(add_key_payload["signature_b64"]) nonce_bytes = bytes.fromhex(nonce_hex) # Raises InvalidSignature if this doesn't verify against the NEW key. new_key.public_key().verify(signature, nonce_bytes) def test_deregisters_old_key_after_successful_add(self, monkeypatch: pytest.MonkeyPatch) -> None: """The old key's key_id is looked up via GET /api/auth/keys/{handle}, matched by fingerprint -- NOT recomputed locally. Recomputing depends on reproducing the exact public_key_b64 encoding the account's original registration used, which silently 404'd in production for accounts registered under an older encoding convention (found running this for real against musehub.ai).""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn from muse.core.keypair import public_key_to_b64url from muse.core.msign import verify_msign_header from muse.core.types import split_pubkey seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() real_old_key_id = "sha256:" + "7" * 64 # deliberately NOT what _compute_key_id would derive list_call = {} delete_call = {} monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "11" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict: list_call["url"] = url list_call["auth_header"] = auth_header return {"keys": [ {"key_id": real_old_key_id, "fingerprint": old_fp}, {"key_id": "sha256:" + "8" * 64, "fingerprint": new_fp}, ]} def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None: delete_call["url"] = url delete_call["auth_header"] = auth_header monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True import urllib.parse assert "gabriel" in list_call["url"] assert urllib.parse.quote(real_old_key_id) in delete_call["url"] assert "gabriel" in delete_call["url"] _, old_pub_b64_bare = split_pubkey(public_key_to_b64url(old_key.public_key())) verified, reason = verify_msign_header( list_call["auth_header"], "GET", list_call["url"], None, old_pub_b64_bare ) assert verified, reason verified, reason = verify_msign_header( delete_call["auth_header"], "DELETE", delete_call["url"], None, old_pub_b64_bare ) assert verified, reason def test_delete_skipped_when_old_key_not_found_in_hub_list(self, monkeypatch: pytest.MonkeyPatch) -> None: """If the hub's key list doesn't contain the old fingerprint at all, deletion is skipped (nothing to delete) rather than guessing a key_id -- still non-fatal.""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() delete_called = [] monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "33" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) monkeypatch.setattr( "muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "8" * 64, "fingerprint": new_fp}]}, ) monkeypatch.setattr( "muse.cli.commands.auth._hub_delete", lambda *a, **kw: delete_called.append(1), ) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True assert delete_called == [] def test_delete_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None: """Old-key deregistration failing must not undo the fact that the new key was already successfully registered -- mirrors `muse auth rotate`.""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "22" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) monkeypatch.setattr( "muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": old_fp}]}, ) def _boom_delete(*a, **kw): raise ConnectionError("hub unreachable for delete") monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _boom_delete) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True def test_lookup_failure_is_non_fatal(self, monkeypatch: pytest.MonkeyPatch) -> None: """If GET /api/auth/keys/{handle} itself fails, deregistration is skipped entirely (non-fatal) -- the new key is already registered regardless.""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "44" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) def _boom_get(*a, **kw): raise ConnectionError("hub unreachable for key list") monkeypatch.setattr("muse.cli.commands.auth._hub_get", _boom_get) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is True def test_missing_challenge_token_fails_closed(self, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "", "is_new_key": True}, ) add_key_called = [] monkeypatch.setattr( "muse.cli.commands.auth._json_post_raw", lambda *a, **kw: add_key_called.append(1) or {}, ) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is False assert add_key_called == [] def test_hub_http_failure_from_challenge_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() def _boom(base_url: str, payload: dict) -> dict: raise ConnectionError("hub unreachable") monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _boom) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is False def test_systemexit_from_add_key_returns_false_not_raise(self, monkeypatch: pytest.MonkeyPatch) -> None: """_json_post_raw raises SystemExit (not a plain Exception) on a real HTTP error -- e.g. the actual HTTP 409 hit in production testing when this shim still called the wrong (fresh-registration) endpoint. A multi-hub live run must not let one hub's HTTP error abort the whole batch.""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn seed, old_key, new_key, old_fp, new_fp, new_path, entry = self._setup() monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "33" * 16, "is_new_key": True}, ) def _boom_post(*a, **kw): raise SystemExit(1) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _boom_post) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", new_fp, new_path, entry) assert ok is False # ============================================================================ # 9. CLI --hub filter # ============================================================================ class TestCliHubFilter: def test_hub_filter_restricts_to_named_hub( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, "staging.musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "b" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke( None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://musehub.ai"] ) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["entries_found"] == 1 assert data["results"][0]["hub_key"] == "musehub.ai" def test_hub_filter_unknown_hub_errors( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke( None, ["migrate", "hub-scoping", "--dry-run", "--json", "--hub", "https://nope.example.com"] ) assert result.exit_code != 0 # ============================================================================ # 10. CLI live run — end-to-end through the real (fixed) registration shim # ============================================================================ class TestCliLiveRunRegistersForReal: def test_live_run_calls_real_challenge_and_verify_with_correct_payloads( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) challenge_calls = [] add_key_calls = [] list_calls = [] delete_calls = [] def _fake_challenge(base_url: str, payload: dict) -> dict: challenge_calls.append((base_url, payload)) return {"challenge_token": "ef" * 16, "is_new_key": True} def _fake_json_post_raw(base_url: str, path: str, payload: dict, extra_headers=None) -> dict: add_key_calls.append((base_url, path, payload, extra_headers)) return {} def _fake_get(url: str, auth_header: str, ssl_ctx=None) -> dict: list_calls.append((url, auth_header)) # The fixture identity entry above has fingerprint "a" * 64 -- the old # (pre-migration) fingerprint the migration will look up by. return {"keys": [{"key_id": "sha256:" + "7" * 64, "fingerprint": "a" * 64}]} def _fake_delete(url: str, auth_header: str, ssl_ctx=None) -> None: delete_calls.append((url, auth_header)) monkeypatch.setattr("muse.cli.commands.auth._post_challenge", _fake_challenge) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _fake_json_post_raw) monkeypatch.setattr("muse.cli.commands.auth._hub_get", _fake_get) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", _fake_delete) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "hub-scoping", "--json"]) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["entries_migrated"] == 1 assert data["results"][0]["hub_registered"] is True assert len(challenge_calls) == 1 assert len(add_key_calls) == 1 challenge_payload = challenge_calls[0][1] _, add_key_path, add_key_payload, add_key_headers = add_key_calls[0] assert challenge_payload["fingerprint"] == data["results"][0]["new_fingerprint"] assert add_key_path == "/api/auth/keys" assert add_key_payload["challenge_token"] == "ef" * 16 assert "public_key_b64" in add_key_payload assert "signature_b64" in add_key_payload assert add_key_headers["Authorization"].startswith("MSign ") # Old key deregistration was attempted, signed by the old key too. assert len(delete_calls) == 1 assert "gabriel" in delete_calls[0][0] assert delete_calls[0][1].startswith("MSign ") import tomllib toml_data = tomllib.loads(identity_file.read_text()) assert toml_data["musehub.ai"]["fingerprint"] == data["results"][0]["new_fingerprint"] # ============================================================================ # 11. Security — secret leakage, terminal injection, and mnemonic safety # # Mirrors test_auth_rotate.py's TestRotateSecurity for the migration command. # ============================================================================ class TestMigrationSecurity: def test_mnemonic_not_in_dry_run_json_output( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner result = CliRunner().invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"]) assert result.exit_code == 0, result.output assert FAKE_MNEMONIC not in (result.output or "") def test_mnemonic_not_in_live_run_json_output( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "55" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": []}) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None) from tests.cli_test_helper import CliRunner result = CliRunner().invoke(None, ["migrate", "hub-scoping", "--json"]) assert result.exit_code == 0, result.output assert FAKE_MNEMONIC not in (result.output or "") def test_no_pem_written_during_migration( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Keys are derived from the mnemonic at sign time -- migration must never write PEM material to disk, matching every other auth command.""" monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "66" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": []}) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None) from tests.cli_test_helper import CliRunner CliRunner().invoke(None, ["migrate", "hub-scoping", "--json"]) keys_dir = dot_muse / "keys" pem_files = list(keys_dir.glob("**/*.pem")) if keys_dir.exists() else [] assert pem_files == [], f"PEM files must not be written during migration: {pem_files}" def test_malicious_add_key_response_size_is_bounded( self, monkeypatch: pytest.MonkeyPatch ) -> None: """The POST /api/auth/keys response is parsed via _json_post_raw, which already enforces the shared 1 MiB response bound -- confirm the migration path actually goes through it rather than a bespoke, unbounded parse.""" from muse.cli.commands.migrate_cmd import _make_hub_register_fn from muse.core.hub_scoping_migration import new_path_for_pre_hub_scoping from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) new_path = new_path_for_pre_hub_scoping(_pre_scoping_path(), "musehub.ai") entry = {"handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64} monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "77" * 16, "is_new_key": True}, ) def _oversized_post_raw(base_url, path, payload, extra_headers=None): import muse.cli.commands.auth as _auth_mod import io class _FakeResp: def read(self, n): return b"x" * (_auth_mod._MAX_RESPONSE_BYTES + 1) def __enter__(self): return self def __exit__(self, *a): return False monkeypatch.setattr( "muse.cli.commands.auth.urllib.request.urlopen", lambda req, timeout=30, context=None: _FakeResp(), ) from muse.cli.commands.auth import _json_post_raw as _real return _real(base_url, path, payload, extra_headers) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", _oversized_post_raw) register_fn = _make_hub_register_fn(seed, json_out=True) ok = register_fn("musehub.ai", "sha256:" + "b" * 64, new_path, entry) assert ok is False, "an oversized hub response must fail closed, not be silently accepted" # ============================================================================ # 12. Performance # # Mirrors test_auth_rotate.py's TestRotatePerformance. # ============================================================================ class TestMigrationPerformance: def test_dry_run_completes_under_500ms( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: import time monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner start = time.perf_counter() result = CliRunner().invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"]) elapsed_ms = (time.perf_counter() - start) * 1000 assert result.exit_code == 0, result.output assert elapsed_ms < 500, f"dry-run took {elapsed_ms:.1f} ms — expected under 500 ms." def test_live_run_with_stubbed_hub_completes_under_500ms( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: import time monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) monkeypatch.setattr( "muse.cli.commands.auth._post_challenge", lambda base_url, payload: {"challenge_token": "88" * 16, "is_new_key": True}, ) monkeypatch.setattr("muse.cli.commands.auth._json_post_raw", lambda *a, **kw: {}) monkeypatch.setattr("muse.cli.commands.auth._hub_get", lambda *a, **kw: {"keys": []}) monkeypatch.setattr("muse.cli.commands.auth._hub_delete", lambda *a, **kw: None) from tests.cli_test_helper import CliRunner start = time.perf_counter() result = CliRunner().invoke(None, ["migrate", "hub-scoping", "--json"]) elapsed_ms = (time.perf_counter() - start) * 1000 assert result.exit_code == 0, result.output assert elapsed_ms < 500, f"live run (stubbed hub) took {elapsed_ms:.1f} ms — expected under 500 ms." # ============================================================================ # 13. Stress # # Mirrors test_auth_rotate.py's TestRotateStress. # ============================================================================ class TestMigrationStress: def test_twenty_hub_entries_migrate_correctly_in_one_run(self) -> None: """20 pre-hub-scoping entries in one identity_map, migrated in a single run: every entry gets a distinct new fingerprint, none collide, and the correct hub_key maps to the correct result -- guards against any cross-contamination between entries processed in the same loop.""" from muse.core.hub_scoping_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { f"hub{i}.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _pre_scoping_path(), "fingerprint": "a" * 64, "algorithm": "ed25519", } for i in range(20) } seed = mnemonic_to_seed(FAKE_MNEMONIC) results = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False ) assert len(results) == 20 fps = {r.new_fingerprint for r in results} assert len(fps) == 20, "all 20 hubs must get distinct migrated fingerprints" for r in results: assert identity_map[r.hub_key]["fingerprint"] == r.new_fingerprint def test_repeated_dry_run_calls_are_stable( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _pre_scoping_path(), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() results = [] for _ in range(10): r = runner.invoke(None, ["migrate", "hub-scoping", "--dry-run", "--json"]) assert r.exit_code == 0, r.output results.append(json.loads(r.output)["results"]) assert all(r == results[0] for r in results), ( "repeated dry-runs against an unchanged identity.toml must be deterministic" )