"""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 # Both entries still get their hd_path/fingerprint updated. assert identity_map["ok.musehub.ai"]["hd_path"] != _pre_scoping_path() assert identity_map["bad.musehub.ai"]["hd_path"] != _pre_scoping_path() 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" # ============================================================================ # 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)