keychain.py
python
sha256:057be43e401106b2b4d877e8028ae67d16a966562d7429707b71a5eaacd5027b
docs(mwp-master): tick all ACs green; mark MWP-7 complete w…
Sonnet 4.6
20 days ago
| 1 | """Secure credential storage via OS keychain. |
| 2 | |
| 3 | BIP39 mnemonics are stored here — never in plaintext TOML files. |
| 4 | |
| 5 | One mnemonic per machine. The same BIP39 mnemonic is the root of all identity |
| 6 | keys across every hub (localhost, staging, prod). Cross-hub replay is prevented |
| 7 | by the host field in the MSign canonical message — not by separate mnemonics. |
| 8 | |
| 9 | The module wraps the ``keyring`` library, which selects the best available |
| 10 | backend automatically: |
| 11 | |
| 12 | macOS → Keychain Services (Secure Enclave on Apple Silicon) |
| 13 | Linux → SecretService (GNOME Keyring / KWallet) |
| 14 | Headless → ``keyrings.alt`` encrypted file (AES + PBKDF2) |
| 15 | |
| 16 | Test / CI isolation |
| 17 | ------------------- |
| 18 | Set ``MUSE_KEYCHAIN_BACKEND=disabled`` to bypass the keychain entirely. |
| 19 | Under this setting all operations are no-ops and :func:`is_available` |
| 20 | returns ``False``. Callers must handle mnemonic absence gracefully (the |
| 21 | mnemonic is ephemeral for the lifetime of the process). |
| 22 | |
| 23 | Key format |
| 24 | ---------- |
| 25 | :: |
| 26 | |
| 27 | service = "muse" |
| 28 | username = "mnemonic" (single global entry — not per-hub) |
| 29 | |
| 30 | Legacy migration |
| 31 | ---------------- |
| 32 | Older versions stored the mnemonic per-hub as ``"{hostname}/mnemonic"``. |
| 33 | :func:`load` transparently promotes any such legacy entry to the global |
| 34 | ``"mnemonic"`` slot on first access and deletes the stale per-hub entry. |
| 35 | |
| 36 | Public API |
| 37 | ---------- |
| 38 | :: |
| 39 | |
| 40 | is_available() -> bool |
| 41 | store(mnemonic) -> bool |
| 42 | load() -> str | None |
| 43 | delete() -> bool |
| 44 | """ |
| 45 | |
| 46 | import logging |
| 47 | import os |
| 48 | import types |
| 49 | |
| 50 | logger = logging.getLogger(__name__) |
| 51 | |
| 52 | _SERVICE = "muse" |
| 53 | _USERNAME = "mnemonic" |
| 54 | |
| 55 | def is_available() -> bool: |
| 56 | """Return ``True`` when a functional keychain backend is active. |
| 57 | |
| 58 | Returns ``False`` when: |
| 59 | - ``MUSE_KEYCHAIN_BACKEND=disabled`` is set (test/CI mode), or |
| 60 | - no keychain backend with priority > 0 is found, or |
| 61 | - the ``keyring`` library is not installed. |
| 62 | |
| 63 | Returns: |
| 64 | ``True`` if credentials can be stored and retrieved. |
| 65 | """ |
| 66 | if os.environ.get("MUSE_KEYCHAIN_BACKEND") == "disabled": |
| 67 | return False |
| 68 | try: |
| 69 | import keyring |
| 70 | backend = keyring.get_keyring() |
| 71 | return getattr(backend, "priority", 0) > 0 |
| 72 | except Exception: |
| 73 | return False |
| 74 | |
| 75 | def store(mnemonic: str) -> bool: |
| 76 | """Store *mnemonic* in the OS keychain as the global machine identity. |
| 77 | |
| 78 | Args: |
| 79 | mnemonic: BIP39 mnemonic phrase to store. |
| 80 | |
| 81 | Returns: |
| 82 | ``True`` on success, ``False`` if the keychain is unavailable or |
| 83 | the store operation fails. |
| 84 | """ |
| 85 | if not is_available(): |
| 86 | return False |
| 87 | try: |
| 88 | import keyring |
| 89 | keyring.set_password(_SERVICE, _USERNAME, mnemonic) |
| 90 | logger.debug("✅ Mnemonic stored in keychain") |
| 91 | return True |
| 92 | except Exception as exc: |
| 93 | logger.warning("⚠️ Could not store mnemonic in keychain: %s", exc) |
| 94 | return False |
| 95 | |
| 96 | def load() -> str | None: |
| 97 | """Load the global mnemonic from the OS keychain. |
| 98 | |
| 99 | Transparently migrates legacy per-hub entries (``"{hostname}/mnemonic"``) |
| 100 | to the global ``"mnemonic"`` slot on first call. |
| 101 | |
| 102 | Returns: |
| 103 | The mnemonic phrase, or ``None`` if not found or keychain unavailable. |
| 104 | """ |
| 105 | if not is_available(): |
| 106 | return None |
| 107 | try: |
| 108 | import keyring |
| 109 | import keyring.errors |
| 110 | |
| 111 | mnemonic = keyring.get_password(_SERVICE, _USERNAME) |
| 112 | if mnemonic is not None: |
| 113 | return mnemonic |
| 114 | |
| 115 | # Legacy migration: scan for per-hub "{hostname}/mnemonic" entries. |
| 116 | # We can't enumerate all keychain entries via the keyring API, so we |
| 117 | # rely on callers to tell us which legacy hostnames to look for via |
| 118 | # the identity file. Fall back to the known common names if identity |
| 119 | # is unavailable. |
| 120 | mnemonic = _migrate_legacy_entry(keyring) |
| 121 | return mnemonic |
| 122 | |
| 123 | except Exception as exc: |
| 124 | logger.warning("⚠️ Could not load mnemonic from keychain: %s", exc) |
| 125 | return None |
| 126 | |
| 127 | def _migrate_legacy_entry(keyring_mod: types.ModuleType) -> str | None: |
| 128 | """Promote a legacy per-hub mnemonic entry to the global slot. |
| 129 | |
| 130 | Scans hostnames recorded in ``~/.muse/identity.toml`` for legacy entries |
| 131 | of the form ``"{hostname}/mnemonic"`` in the keychain. The first one found |
| 132 | is promoted to ``"mnemonic"`` and all per-hub entries are deleted. |
| 133 | |
| 134 | Returns the migrated mnemonic, or ``None`` if none found. |
| 135 | """ |
| 136 | import keyring as kr |
| 137 | import keyring.errors |
| 138 | |
| 139 | # Collect candidate hostnames from identity.toml (best-effort). |
| 140 | hostnames: list[str] = [] |
| 141 | try: |
| 142 | from muse.core.identity import list_all_identities |
| 143 | for key in list_all_identities(): |
| 144 | # Keys are bare hostnames or "hostname#agent_id" |
| 145 | hostname = key.split("#")[0] |
| 146 | if hostname and hostname not in hostnames: |
| 147 | hostnames.append(hostname) |
| 148 | except Exception: |
| 149 | pass |
| 150 | |
| 151 | migrated: str | None = None |
| 152 | legacy_keys_found: list[str] = [] |
| 153 | |
| 154 | for hostname in hostnames: |
| 155 | legacy_username = f"{hostname}/mnemonic" |
| 156 | value = kr.get_password(_SERVICE, legacy_username) |
| 157 | if value is not None: |
| 158 | legacy_keys_found.append(legacy_username) |
| 159 | if migrated is None: |
| 160 | migrated = value |
| 161 | |
| 162 | if migrated is None: |
| 163 | return None |
| 164 | |
| 165 | # Write the global entry. |
| 166 | try: |
| 167 | kr.set_password(_SERVICE, _USERNAME, migrated) |
| 168 | logger.info("✅ Mnemonic migrated from per-hub keychain entry to global slot") |
| 169 | except Exception as exc: |
| 170 | logger.warning("⚠️ Could not write global mnemonic during migration: %s", exc) |
| 171 | return migrated # still return the value even if we couldn't promote it |
| 172 | |
| 173 | # Delete all legacy per-hub entries. |
| 174 | for legacy_username in legacy_keys_found: |
| 175 | try: |
| 176 | kr.delete_password(_SERVICE, legacy_username) |
| 177 | except Exception: |
| 178 | pass |
| 179 | |
| 180 | return migrated |
| 181 | |
| 182 | def delete() -> bool: |
| 183 | """Delete the global mnemonic from the OS keychain. |
| 184 | |
| 185 | Returns: |
| 186 | ``True`` if the entry was deleted, ``False`` if it was not found or |
| 187 | the keychain is unavailable. |
| 188 | """ |
| 189 | if not is_available(): |
| 190 | return False |
| 191 | try: |
| 192 | import keyring |
| 193 | import keyring.errors |
| 194 | keyring.delete_password(_SERVICE, _USERNAME) |
| 195 | logger.debug("✅ Mnemonic deleted from keychain") |
| 196 | return True |
| 197 | except Exception as exc: |
| 198 | logger.debug("keychain delete: %s", exc) |
| 199 | return False |
File History
1 commit
sha256:057be43e401106b2b4d877e8028ae67d16a966562d7429707b71a5eaacd5027b
docs(mwp-master): tick all ACs green; mark MWP-7 complete w…
Sonnet 4.6
20 days ago