keypair.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
| 1 | """Ed25519 keypair management for Muse CLI authentication. |
| 2 | |
| 3 | Private keys are stored in ``~/.muse/keys/``, one PEM file per hub hostname |
| 4 | (or per ``hostname__agent_id`` for agent-specific keys). |
| 5 | The private key is PKCS8-encoded PEM (no passphrase) with strict 0o600 |
| 6 | permissions, written atomically to prevent partial-write corruption. |
| 7 | |
| 8 | Agent key paths |
| 9 | --------------- |
| 10 | Human: ``~/.muse/keys/localhost_10003.pem`` |
| 11 | Agent: ``~/.muse/keys/localhost_10003__agentception-abc123.pem`` |
| 12 | |
| 13 | The double-underscore ``__`` separates hostname from agent_id in the filename. |
| 14 | Forward-slashes and colons in the hostname are replaced with ``_`` as usual; |
| 15 | the agent_id portion is appended verbatim (it is a safe handle string). |
| 16 | |
| 17 | Public key encoding |
| 18 | -------------------- |
| 19 | Public keys are transmitted as URL-safe base64 (no padding) of the raw 32-byte |
| 20 | Ed25519 public key material — identical to the format expected by MuseHub's |
| 21 | ``/api/auth/challenge`` and ``/api/auth/verify`` endpoints. |
| 22 | |
| 23 | Fingerprint |
| 24 | ----------- |
| 25 | SHA-256 hex digest of the raw 32 public key bytes, lowercase, no separators. |
| 26 | This is the stable identifier used by MuseHub to look up a registered key. |
| 27 | |
| 28 | Security properties |
| 29 | ------------------- |
| 30 | - Private keys are never logged, printed, or included in exception messages. |
| 31 | - Key files are written with ``fchmod(0o600)`` *before* any data is written, |
| 32 | eliminating the TOCTOU window that ``write_text()`` + ``chmod()`` creates. |
| 33 | - Writes are atomic: data goes to a temp file in the same directory, then |
| 34 | ``os.replace()`` renames it over the target. |
| 35 | - Symlink guard: if the target path is already a symlink, write is refused. |
| 36 | - The ``~/.muse/keys/`` directory is created with mode 0o700 (user-only). |
| 37 | """ |
| 38 | from __future__ import annotations |
| 39 | |
| 40 | import base64 |
| 41 | import hashlib |
| 42 | import logging |
| 43 | import os |
| 44 | import pathlib |
| 45 | import stat |
| 46 | import tempfile |
| 47 | |
| 48 | from cryptography.hazmat.primitives.asymmetric.ed25519 import ( |
| 49 | Ed25519PrivateKey, |
| 50 | Ed25519PublicKey, |
| 51 | ) |
| 52 | from cryptography.hazmat.primitives.serialization import ( |
| 53 | Encoding, |
| 54 | NoEncryption, |
| 55 | PrivateFormat, |
| 56 | PublicFormat, |
| 57 | ) |
| 58 | |
| 59 | logger = logging.getLogger(__name__) |
| 60 | |
| 61 | _KEYS_DIR = pathlib.Path.home() / ".muse" / "keys" |
| 62 | |
| 63 | |
| 64 | # --------------------------------------------------------------------------- |
| 65 | # Internal helpers |
| 66 | # --------------------------------------------------------------------------- |
| 67 | |
| 68 | |
| 69 | def _hostname_key(hostname: str) -> str: |
| 70 | """Sanitise *hostname* to a safe filename component. |
| 71 | |
| 72 | Replaces ``:`` and ``/`` with ``_`` so that ``localhost:10003`` becomes a |
| 73 | valid filename without path-traversal risk. |
| 74 | """ |
| 75 | return hostname.replace(":", "_").replace("/", "_").lower() |
| 76 | |
| 77 | |
| 78 | def _key_path(hostname: str, agent_id: str | None = None) -> pathlib.Path: |
| 79 | """Return the path to the PKCS8 PEM private key file for *hostname*. |
| 80 | |
| 81 | Agent keys use a ``{hostname}__{agent_id}.pem`` filename so that human |
| 82 | and agent keys never collide in ``~/.muse/keys/``. |
| 83 | """ |
| 84 | base = _hostname_key(hostname) |
| 85 | if agent_id: |
| 86 | return _KEYS_DIR / f"{base}__{agent_id}.pem" |
| 87 | return _KEYS_DIR / f"{base}.pem" |
| 88 | |
| 89 | |
| 90 | def _ensure_keys_dir() -> None: |
| 91 | """Create ``~/.muse/keys/`` with 0o700 permissions if it does not exist.""" |
| 92 | _KEYS_DIR.mkdir(parents=True, exist_ok=True) |
| 93 | try: |
| 94 | os.chmod(_KEYS_DIR, stat.S_IRWXU) # 0o700 |
| 95 | except OSError as exc: |
| 96 | logger.warning("⚠️ Could not set permissions on %s: %s", _KEYS_DIR, exc) |
| 97 | |
| 98 | |
| 99 | def _write_private_key_pem(path: pathlib.Path, private_key: Ed25519PrivateKey) -> None: |
| 100 | """Write *private_key* to *path* as an unencrypted PKCS8 PEM file. |
| 101 | |
| 102 | Security guarantees: |
| 103 | - Refuses to write if *path* is a symlink. |
| 104 | - Sets 0o600 via ``fchmod`` *before* any data is written. |
| 105 | - Uses ``os.replace()`` for an atomic rename from a temp file. |
| 106 | """ |
| 107 | if path.is_symlink(): |
| 108 | raise OSError( |
| 109 | f"Security: {path} is a symlink. " |
| 110 | "Refusing to write a private key to a symlink target." |
| 111 | ) |
| 112 | |
| 113 | pem_bytes: bytes = private_key.private_bytes( |
| 114 | encoding=Encoding.PEM, |
| 115 | format=PrivateFormat.PKCS8, |
| 116 | encryption_algorithm=NoEncryption(), |
| 117 | ) |
| 118 | |
| 119 | fd, tmp_path_str = tempfile.mkstemp(dir=path.parent, prefix=".key-tmp-") |
| 120 | tmp_path = pathlib.Path(tmp_path_str) |
| 121 | try: |
| 122 | os.fchmod(fd, stat.S_IRUSR | stat.S_IWUSR) # 0o600 before any data |
| 123 | with os.fdopen(fd, "wb") as fh: |
| 124 | fh.write(pem_bytes) |
| 125 | os.replace(tmp_path, path) |
| 126 | except Exception: |
| 127 | try: |
| 128 | tmp_path.unlink(missing_ok=True) |
| 129 | except OSError: |
| 130 | pass |
| 131 | raise |
| 132 | |
| 133 | |
| 134 | # --------------------------------------------------------------------------- |
| 135 | # Public key encoding helpers |
| 136 | # --------------------------------------------------------------------------- |
| 137 | |
| 138 | |
| 139 | def public_key_to_b64url(public_key: Ed25519PublicKey) -> str: |
| 140 | """Return the URL-safe base64 (no padding) encoding of the raw public key bytes.""" |
| 141 | raw: bytes = public_key.public_bytes(Encoding.Raw, PublicFormat.Raw) |
| 142 | return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") |
| 143 | |
| 144 | |
| 145 | def public_key_fingerprint(public_key: Ed25519PublicKey) -> str: |
| 146 | """Return the SHA-256 hex fingerprint (64 lowercase hex chars) of the raw public key.""" |
| 147 | raw: bytes = public_key.public_bytes(Encoding.Raw, PublicFormat.Raw) |
| 148 | return hashlib.sha256(raw).hexdigest() |
| 149 | |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # Public API |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | |
| 156 | def generate_keypair(hostname: str, agent_id: str | None = None) -> tuple[str, str]: |
| 157 | """Generate a new Ed25519 keypair and persist the private key to disk. |
| 158 | |
| 159 | Human keys are written to ``~/.muse/keys/{hostname}.pem`` (0o600). |
| 160 | Agent keys are written to ``~/.muse/keys/{hostname}__{agent_id}.pem``. |
| 161 | |
| 162 | If a key already exists at the target path it is overwritten — callers |
| 163 | should warn the user before invoking this function if the key exists. |
| 164 | |
| 165 | Args: |
| 166 | hostname: Normalised hub hostname (e.g. ``"localhost:10003"``). |
| 167 | agent_id: Optional agent handle. When supplied the key is stored |
| 168 | under the agent-specific filename. |
| 169 | |
| 170 | Returns: |
| 171 | ``(public_key_b64url, fingerprint)`` — both are safe to log / display. |
| 172 | The private key is NEVER returned; it lives only on disk. |
| 173 | """ |
| 174 | _ensure_keys_dir() |
| 175 | private_key = Ed25519PrivateKey.generate() |
| 176 | public_key = private_key.public_key() |
| 177 | |
| 178 | path = _key_path(hostname, agent_id) |
| 179 | _write_private_key_pem(path, private_key) |
| 180 | logger.info("✅ Ed25519 private key written to %s", path) |
| 181 | |
| 182 | return public_key_to_b64url(public_key), public_key_fingerprint(public_key) |
| 183 | |
| 184 | |
| 185 | def load_private_key(hostname: str, agent_id: str | None = None) -> Ed25519PrivateKey | None: |
| 186 | """Load the Ed25519 private key for *hostname* (and optional *agent_id*) from disk. |
| 187 | |
| 188 | Args: |
| 189 | hostname: Normalised hub hostname. |
| 190 | agent_id: Optional agent handle — loads the agent-specific key file. |
| 191 | |
| 192 | Returns: |
| 193 | :class:`Ed25519PrivateKey` if a key file exists, else ``None``. |
| 194 | Returns ``None`` (not an exception) when no key has been generated |
| 195 | yet — callers should prompt the user to run ``muse auth keygen``. |
| 196 | """ |
| 197 | from cryptography.hazmat.primitives.serialization import load_pem_private_key |
| 198 | |
| 199 | path = _key_path(hostname, agent_id) |
| 200 | if not path.is_file(): |
| 201 | return None |
| 202 | try: |
| 203 | pem_bytes = path.read_bytes() |
| 204 | key = load_pem_private_key(pem_bytes, password=None) |
| 205 | except Exception as exc: |
| 206 | logger.warning("⚠️ Could not load private key from %s (%s)", path, type(exc).__name__) |
| 207 | return None |
| 208 | if not isinstance(key, Ed25519PrivateKey): |
| 209 | logger.warning("⚠️ Key at %s is not an Ed25519 key", path) |
| 210 | return None |
| 211 | return key |
| 212 | |
| 213 | |
| 214 | def sign_bytes(private_key: Ed25519PrivateKey, data: bytes) -> str: |
| 215 | """Sign *data* with *private_key* and return URL-safe base64 (no padding). |
| 216 | |
| 217 | Args: |
| 218 | private_key: Ed25519 private key. |
| 219 | data: Raw bytes to sign (the nonce bytes from the challenge token). |
| 220 | |
| 221 | Returns: |
| 222 | URL-safe base64 (no padding) of the 64-byte signature. |
| 223 | """ |
| 224 | signature: bytes = private_key.sign(data) |
| 225 | return base64.urlsafe_b64encode(signature).rstrip(b"=").decode("ascii") |
| 226 | |
| 227 | |
| 228 | def key_path_for(hostname: str, agent_id: str | None = None) -> pathlib.Path: |
| 229 | """Return the expected private key path for *hostname* (may not exist). |
| 230 | |
| 231 | Args: |
| 232 | hostname: Normalised hub hostname. |
| 233 | agent_id: Optional agent handle — returns the agent-specific key path. |
| 234 | """ |
| 235 | return _key_path(hostname, agent_id) |
| 236 | |
| 237 | |
| 238 | def load_private_key_from_pem(pem_bytes: bytes) -> Ed25519PrivateKey | None: |
| 239 | """Load an Ed25519 private key from raw PEM bytes (e.g. from MUSE_AGENT_KEY env). |
| 240 | |
| 241 | Returns ``None`` if the bytes cannot be decoded or are not Ed25519. |
| 242 | The bytes are never logged. |
| 243 | """ |
| 244 | from cryptography.hazmat.primitives.serialization import load_pem_private_key |
| 245 | |
| 246 | try: |
| 247 | key = load_pem_private_key(pem_bytes, password=None) |
| 248 | except Exception as exc: |
| 249 | logger.warning("⚠️ Could not load private key from PEM bytes (%s)", type(exc).__name__) |
| 250 | return None |
| 251 | if not isinstance(key, Ed25519PrivateKey): |
| 252 | logger.warning("⚠️ PEM key is not Ed25519") |
| 253 | return None |
| 254 | return key |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
33 days ago