"""Tests for muse.core.keychain — OS keychain integration — Tier 2. The keychain module is the only place mnemonics are stored at rest. Plaintext TOML storage of mnemonics is permanently retired. One mnemonic per machine, not one per hub. The same BIP39 mnemonic is the root of all identity keys across every hub (localhost, staging, prod). Cross-hub replay is prevented by the host field in the MSign canonical message, not by using separate mnemonics. Coverage -------- I keychain module API — single global mnemonic (no hub_url) I1 store(mnemonic) → load() round-trip returns the stored phrase I2 load() returns None when no entry exists I3 delete() removes the entry; load() returns None afterward I4 delete() on missing entry returns False without raising I5 is_available returns False when MUSE_KEYCHAIN_BACKEND=disabled I6 load() returns the same mnemonic regardless of which hub is queried I7 store() is idempotent — calling it twice with the same value is safe I8 mnemonic stored for hub A is readable without specifying a hub URL II identity.toml never contains mnemonic II1 save_identity with mnemonic kwarg stores in keychain, not TOML II2 TOML written by save_identity has no "mnemonic" key II3 load_identity retrieves mnemonic from keychain, not TOML II4 identity TOML has no key_source field (derivation is always HD) III keygen stores mnemonic in keychain III1 muse auth keygen --json stdout has no "mnemonic" key III2 identity.toml written after keygen has no mnemonic field III3 keychain holds the mnemonic after keygen IV keychain disabled (MUSE_KEYCHAIN_BACKEND=disabled) IV1 is_available() is False IV2 store() returns False without raising IV3 load() returns None without raising IV4 muse auth keygen still succeeds (mnemonic is ephemeral) V keychain unavailable — operator must be warned (CRITICAL-1) V1 warns when keychain is unavailable for non-intentional reason V2 silent when MUSE_KEYCHAIN_BACKEND=disabled (intentional CI mode) VI legacy per-hub mnemonic migration VI1 load() promotes a legacy "{hostname}/mnemonic" entry to "mnemonic" VI2 after migration the legacy entry is deleted from the keychain VI3 migration is idempotent — running load() twice does not corrupt state VI4 if both global and legacy entries exist, global wins (no overwrite) VI5 two legacy entries for different hubs both migrate to the same slot """ from __future__ import annotations import json import os import pathlib import pytest try: import tomllib except ModuleNotFoundError: import tomli as tomllib # type: ignore[no-reuse-def] from tests.cli_test_helper import CliRunner from muse.core.paths import muse_dir cli = None runner = CliRunner() _TEST_HUB = "https://localhost:1337" _TEST_HOSTNAME = "localhost:1337" _TEST_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon abandon abandon " "abandon abandon abandon about" ) # Capture the real keychain implementations at import time — before any # autouse fixture patches them. The keychain_in_memory fixture restores # these so tests that exercise internal keychain logic (migration, etc.) # call through the real code rather than the conftest's in-memory shim. import muse.core.keychain as _kc_module _REAL_KC_STORE = _kc_module.store _REAL_KC_LOAD = _kc_module.load _REAL_KC_DELETE = _kc_module.delete _REAL_KC_IS_AVAILABLE = _kc_module.is_available # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def keychain_in_memory(monkeypatch: pytest.MonkeyPatch) -> dict[tuple[str, str], str]: """Patch keyring to use an in-memory dict as the backend. Returns the dict so tests can inspect it directly. Also restores the real muse.core.keychain functions (which the autouse conftest fixture replaces with a simpler shim) so that tests in this file exercise the actual store/load/delete/migration code paths. """ store: dict[tuple[str, str], str] = {} import keyring monkeypatch.setattr(keyring, "set_password", lambda svc, usr, pwd: store.__setitem__((svc, usr), pwd)) monkeypatch.setattr(keyring, "get_password", lambda svc, usr: store.get((svc, usr))) import keyring.errors def _delete(svc: str, usr: str) -> None: if (svc, usr) not in store: raise keyring.errors.PasswordDeleteError("not found") del store[(svc, usr)] monkeypatch.setattr(keyring, "delete_password", _delete) # Restore the real module-level functions so these tests exercise the # actual keychain code (including migration logic), not the conftest shim. import muse.core.keychain as kc_mod monkeypatch.setattr(kc_mod, "store", _REAL_KC_STORE) monkeypatch.setattr(kc_mod, "load", _REAL_KC_LOAD) monkeypatch.setattr(kc_mod, "delete", _REAL_KC_DELETE) monkeypatch.setattr(kc_mod, "is_available", lambda: True) monkeypatch.delenv("MUSE_KEYCHAIN_BACKEND", raising=False) return store # type: ignore[return-value] @pytest.fixture() def isolated_identity(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: fake_dir = tmp_path / "dot_muse" fake_dir.mkdir() monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_dir) monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_dir / "identity.toml") return fake_dir @pytest.fixture() def isolated_keys(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: keys_dir = tmp_path / "keys" keys_dir.mkdir() monkeypatch.setattr("muse.core.keypair._KEYS_DIR", keys_dir) return keys_dir @pytest.fixture() def repo_with_hub(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: dot_muse = muse_dir(tmp_path) dot_muse.mkdir() (dot_muse / "HEAD").write_text("ref: refs/heads/main\n") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "objects").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "snapshots").mkdir() (dot_muse / "config.toml").write_text(f'[hub]\nurl = "{_TEST_HUB}"\n') monkeypatch.chdir(tmp_path) return tmp_path # --------------------------------------------------------------------------- # I keychain module API — single global mnemonic (no hub_url) # --------------------------------------------------------------------------- class TestKeychainApiI: def test_I1_store_load_roundtrip( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I1: store(mnemonic) and load() round-trip — no hub URL required.""" from muse.core.keychain import store, load assert store(_TEST_MNEMONIC) assert load() == _TEST_MNEMONIC def test_I2_load_missing_returns_none( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I2: load() returns None when nothing has been stored.""" from muse.core.keychain import load assert load() is None def test_I3_delete_removes_entry( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I3: delete() removes the global entry; subsequent load() returns None.""" from muse.core.keychain import store, load, delete store(_TEST_MNEMONIC) assert delete() assert load() is None def test_I4_delete_missing_returns_false( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I4: delete() on empty keychain returns False without raising.""" from muse.core.keychain import delete assert not delete() def test_I5_disabled_backend_not_available( self, monkeypatch: pytest.MonkeyPatch ) -> None: """I5: is_available() is False when MUSE_KEYCHAIN_BACKEND=disabled.""" monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core import keychain import importlib importlib.reload(keychain) assert not keychain.is_available() def test_I6_load_is_hub_agnostic( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I6: the same mnemonic is returned no matter which hub context is active.""" from muse.core.keychain import store, load store(_TEST_MNEMONIC) # load() takes no hub_url — mnemonic is global assert load() == _TEST_MNEMONIC assert load() == _TEST_MNEMONIC # idempotent def test_I7_store_is_idempotent( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I7: calling store() twice with the same value does not corrupt state.""" from muse.core.keychain import store, load assert store(_TEST_MNEMONIC) assert store(_TEST_MNEMONIC) assert load() == _TEST_MNEMONIC def test_I8_single_keychain_entry( self, keychain_in_memory: dict[tuple[str, str], str] ) -> None: """I8: exactly one keychain entry exists after store() — no per-hub duplicates.""" from muse.core.keychain import store store(_TEST_MNEMONIC) # Only one entry in the entire in-memory store assert len(keychain_in_memory) == 1 # And its username is the global constant, not a hostname-scoped key (service, username), value = next(iter(keychain_in_memory.items())) assert service == "muse" assert "/" not in username, ( f"Keychain username '{username}' looks per-hub — expected a global key with no '/'" ) assert value == _TEST_MNEMONIC # --------------------------------------------------------------------------- # II identity.toml never contains mnemonic # --------------------------------------------------------------------------- class TestIdentityNoMnemonicII: def test_II1_save_stores_mnemonic_in_keychain( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """save_identity with mnemonic puts it in keychain, not TOML.""" from muse.core.identity import save_identity, IdentityEntry entry: IdentityEntry = { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, } save_identity(_TEST_HUB, entry, mnemonic=_TEST_MNEMONIC) # Keychain has the mnemonic from muse.core.keychain import load as kc_load assert kc_load() == _TEST_MNEMONIC def test_II2_toml_has_no_mnemonic_key( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """The TOML file written by save_identity must not contain 'mnemonic'.""" from muse.core.identity import save_identity, IdentityEntry entry: IdentityEntry = { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, } save_identity(_TEST_HUB, entry, mnemonic=_TEST_MNEMONIC) toml_text = (isolated_identity / "identity.toml").read_text() assert "mnemonic" not in toml_text.lower(), ( f"'mnemonic' found in TOML:\n{toml_text}" ) def test_II3_load_retrieves_mnemonic_from_keychain( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """load_identity fetches the mnemonic from keychain and injects it.""" from muse.core.identity import save_identity, load_identity, IdentityEntry entry: IdentityEntry = { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, } save_identity(_TEST_HUB, entry, mnemonic=_TEST_MNEMONIC) loaded = load_identity(_TEST_HUB) assert loaded is not None assert loaded.get("mnemonic") == _TEST_MNEMONIC def test_II4_toml_has_no_key_source_field( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """TOML must not contain a key_source field — derivation method is implied.""" from muse.core.identity import save_identity, IdentityEntry entry: IdentityEntry = { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, } save_identity(_TEST_HUB, entry, mnemonic=_TEST_MNEMONIC) toml_text = (isolated_identity / "identity.toml").read_text() assert "key_source" not in toml_text # --------------------------------------------------------------------------- # III keygen stores mnemonic in keychain # --------------------------------------------------------------------------- class TestKeygenUsesKeychainIII: def test_III1_keygen_json_stdout_no_mnemonic( self, isolated_identity: pathlib.Path, isolated_keys: pathlib.Path, repo_with_hub: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], monkeypatch: pytest.MonkeyPatch, ) -> None: """III1: muse auth keygen --json stdout must not contain 'mnemonic'.""" from muse.core import bip39 as bip39_mod monkeypatch.setattr(bip39_mod, "generate_mnemonic", lambda **kw: _TEST_MNEMONIC) result = runner.invoke(cli, ["auth", "keygen", "--hub", _TEST_HUB, "--json"]) assert result.exit_code == 0, f"keygen failed:\n{result.output}" json_lines = [ln for ln in result.stdout.splitlines() if ln.strip().startswith("{")] assert json_lines, "No JSON output found" for line in json_lines: data = json.loads(line) assert "mnemonic" not in data, f"'mnemonic' key in JSON output: {data}" def test_III2_keygen_toml_has_no_mnemonic( self, isolated_identity: pathlib.Path, isolated_keys: pathlib.Path, repo_with_hub: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], monkeypatch: pytest.MonkeyPatch, ) -> None: """III2: identity.toml after keygen must not have mnemonic in plaintext.""" from muse.core import bip39 as bip39_mod monkeypatch.setattr(bip39_mod, "generate_mnemonic", lambda **kw: _TEST_MNEMONIC) runner.invoke(cli, ["auth", "keygen", "--hub", _TEST_HUB, "--json"]) toml_file = isolated_identity / "identity.toml" assert toml_file.exists(), "identity.toml not created" content = toml_file.read_text() assert "mnemonic" not in content.lower(), f"mnemonic in TOML:\n{content}" def test_III3_keychain_holds_mnemonic_after_keygen( self, isolated_identity: pathlib.Path, isolated_keys: pathlib.Path, repo_with_hub: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], monkeypatch: pytest.MonkeyPatch, ) -> None: """III3: keychain holds the mnemonic globally after keygen (no hub scoping).""" from muse.core import bip39 as bip39_mod monkeypatch.setattr(bip39_mod, "generate_mnemonic", lambda **kw: _TEST_MNEMONIC) runner.invoke(cli, ["auth", "keygen", "--hub", _TEST_HUB, "--json"]) from muse.core.keychain import load as kc_load stored = kc_load() assert stored == _TEST_MNEMONIC, f"Keychain does not have mnemonic, got: {stored!r}" # --------------------------------------------------------------------------- # IV keychain disabled # --------------------------------------------------------------------------- class TestKeychainDisabledIV: def test_IV1_is_available_false(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core import keychain import importlib importlib.reload(keychain) assert not keychain.is_available() def test_IV2_store_returns_false(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core import keychain import importlib importlib.reload(keychain) assert not keychain.store(_TEST_MNEMONIC) def test_IV3_load_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core import keychain import importlib importlib.reload(keychain) assert keychain.load() is None def test_IV4_keygen_succeeds_without_keychain( self, isolated_identity: pathlib.Path, isolated_keys: pathlib.Path, repo_with_hub: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV4: keygen still works when keychain is disabled (mnemonic is ephemeral).""" monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core import bip39 as bip39_mod monkeypatch.setattr(bip39_mod, "generate_mnemonic", lambda **kw: _TEST_MNEMONIC) result = runner.invoke(cli, ["auth", "keygen", "--hub", _TEST_HUB, "--json"]) assert result.exit_code == 0, f"keygen failed with disabled keychain:\n{result.output}" # --------------------------------------------------------------------------- # V keychain unavailable — operator must be warned (CRITICAL-1) # --------------------------------------------------------------------------- class TestKeychainUnavailableWarnsV: """V When the keychain is truly unavailable (not intentionally disabled), save_identity must warn the operator that the mnemonic is ephemeral. MUSE_KEYCHAIN_BACKEND=disabled is CI/test mode and must stay silent. Any other cause of is_available()==False is an operational failure and demands a log.warning so the operator knows their root key is not persisted. """ _entry = { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, } def test_V1_warns_when_keychain_unavailable( self, isolated_identity: pathlib.Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: """V1: save_identity logs a warning when keychain is unavailable for a non-intentional reason (no backend, library not installed, etc.). Simulate: is_available() returns False but MUSE_KEYCHAIN_BACKEND is not set. """ import logging from unittest.mock import patch from muse.core import keychain as kc_mod from muse.core.identity import save_identity monkeypatch.delenv("MUSE_KEYCHAIN_BACKEND", raising=False) with patch.object(kc_mod, "is_available", return_value=False): with caplog.at_level(logging.WARNING, logger="muse.core.identity"): save_identity(_TEST_HUB, self._entry, mnemonic=_TEST_MNEMONIC) # type: ignore[arg-type] warning_messages = [ r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING ] assert warning_messages, ( "Expected a warning about unavailable keychain — got none.\n" f"All log records: {[r.getMessage() for r in caplog.records]}" ) combined = " ".join(warning_messages).lower() assert "keychain" in combined or "ephemeral" in combined, ( f"Warning must mention 'keychain' or 'ephemeral': {warning_messages}" ) def test_V2_silent_when_keychain_intentionally_disabled( self, isolated_identity: pathlib.Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: """V2: no keychain warning when MUSE_KEYCHAIN_BACKEND=disabled (CI/test mode). The disabled env var signals intentional ephemeral operation — the operator has opted out of keychain storage on purpose, so no warning should fire. """ import logging from muse.core.identity import save_identity monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") with caplog.at_level(logging.WARNING, logger="muse.core.identity"): save_identity(_TEST_HUB, self._entry, mnemonic=_TEST_MNEMONIC) # type: ignore[arg-type] keychain_warnings = [ r.getMessage() for r in caplog.records if r.levelno >= logging.WARNING and ("keychain" in r.getMessage().lower() or "ephemeral" in r.getMessage().lower()) ] assert not keychain_warnings, ( f"Unexpected keychain warning in intentional CI/disabled mode: {keychain_warnings}" ) # --------------------------------------------------------------------------- # VI legacy per-hub mnemonic migration # --------------------------------------------------------------------------- _STAGING_HUB = "https://staging.musehub.ai" _STAGING_HOSTNAME = "staging.musehub.ai" _LEGACY_USERNAME_LOCALHOST = "localhost:1337/mnemonic" _LEGACY_USERNAME_STAGING = "staging.musehub.ai/mnemonic" _GLOBAL_USERNAME = "mnemonic" def _seed_identity_toml(isolated_identity: pathlib.Path, *hostnames: str) -> None: """Write a minimal identity.toml so the migration scanner finds the hostnames.""" lines = [] for h in hostnames: lines.append(f'["{h}"]') lines.append('type = "human"') lines.append('handle = "gabriel"') lines.append('algorithm = "ed25519"') lines.append(f'fingerprint = "{"a" * 64}"') lines.append('hd_path = "m/1075233755\'/0\'/0\'/0\'/0\'/0\'"') lines.append("") (isolated_identity / "identity.toml").write_text("\n".join(lines)) class TestLegacyMigrationVI: """VI Transparent migration from per-hub keychain entries to a single global entry. Users who ran older versions of muse have entries keyed as "{hostname}/mnemonic" in the keychain. load() must silently promote one of these to the global "mnemonic" slot and delete the legacy entry. Migration rules: - Any "{hostname}/mnemonic" entry found → promoted to "mnemonic" - The legacy entry is deleted after promotion - If global "mnemonic" already exists, legacy entries are only cleaned up (no overwrite — the operator explicitly stored a global mnemonic already) - Migration is idempotent """ def test_VI1_load_promotes_legacy_entry( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI1: load() finds a legacy per-hub entry and returns its mnemonic.""" _seed_identity_toml(isolated_identity, _TEST_HOSTNAME) keychain_in_memory[("muse", _LEGACY_USERNAME_LOCALHOST)] = _TEST_MNEMONIC from muse.core.keychain import load result = load() assert result == _TEST_MNEMONIC, ( f"Expected legacy mnemonic to be returned, got: {result!r}" ) def test_VI2_migration_deletes_legacy_entry( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI2: after load() migrates a legacy entry, the old key is gone.""" _seed_identity_toml(isolated_identity, _TEST_HOSTNAME) keychain_in_memory[("muse", _LEGACY_USERNAME_LOCALHOST)] = _TEST_MNEMONIC from muse.core.keychain import load load() assert ("muse", _LEGACY_USERNAME_LOCALHOST) not in keychain_in_memory, ( "Legacy per-hub entry should have been deleted after migration" ) def test_VI3_migration_writes_global_entry( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI3: after load() migrates a legacy entry, the global key is written.""" _seed_identity_toml(isolated_identity, _TEST_HOSTNAME) keychain_in_memory[("muse", _LEGACY_USERNAME_LOCALHOST)] = _TEST_MNEMONIC from muse.core.keychain import load load() assert ("muse", _GLOBAL_USERNAME) in keychain_in_memory, ( "Global 'mnemonic' key should exist after migration" ) assert keychain_in_memory[("muse", _GLOBAL_USERNAME)] == _TEST_MNEMONIC def test_VI4_migration_is_idempotent( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI4: calling load() twice with a legacy entry doesn't corrupt state.""" _seed_identity_toml(isolated_identity, _TEST_HOSTNAME) keychain_in_memory[("muse", _LEGACY_USERNAME_LOCALHOST)] = _TEST_MNEMONIC from muse.core.keychain import load first = load() second = load() assert first == _TEST_MNEMONIC assert second == _TEST_MNEMONIC assert len(keychain_in_memory) == 1 assert ("muse", _GLOBAL_USERNAME) in keychain_in_memory def test_VI5_global_wins_over_legacy( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI5: if both global and legacy entries exist, global is returned unchanged.""" _seed_identity_toml(isolated_identity, _STAGING_HOSTNAME) other_mnemonic = "zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong" keychain_in_memory[("muse", _GLOBAL_USERNAME)] = _TEST_MNEMONIC keychain_in_memory[("muse", _LEGACY_USERNAME_STAGING)] = other_mnemonic from muse.core.keychain import load result = load() assert result == _TEST_MNEMONIC, ( "Global entry should win — legacy entry must not overwrite it" ) def test_VI6_two_legacy_hubs_migrate_to_one_slot( self, isolated_identity: pathlib.Path, keychain_in_memory: dict[tuple[str, str], str], ) -> None: """VI6: two legacy per-hub entries for different hubs both collapse to one global entry.""" _seed_identity_toml(isolated_identity, _TEST_HOSTNAME, _STAGING_HOSTNAME) keychain_in_memory[("muse", _LEGACY_USERNAME_LOCALHOST)] = _TEST_MNEMONIC keychain_in_memory[("muse", _LEGACY_USERNAME_STAGING)] = _TEST_MNEMONIC from muse.core.keychain import load result = load() assert result == _TEST_MNEMONIC remaining_keys = {usr for (svc, usr) in keychain_in_memory if svc == "muse"} assert remaining_keys == {_GLOBAL_USERNAME}, ( f"Only global key should remain after migration, got: {remaining_keys}" )