"""muse auth — identity management. Muse has two primary user types: **humans** and **agents**. Both are first-class identities authenticated via Ed25519 key-pair challenge-response. This command manages the identity lifecycle: keygen, register, whoami, logout. Why not ``muse config set`` for credentials? --------------------------------------------- Credentials belong to the machine, not the repository. Storing credentials inside ``.muse/config.toml`` means they could be committed to version control, shared across repos accidentally, or tied to a single repo when the identity is global. Instead: - Credentials live in ``~/.muse/identity.toml`` (mode 0o600, never read by the snapshot engine). - ``config.toml`` records *where* the hub is (``[hub] url``), not *who you are*. - This command owns the identity lifecycle: keygen, register, whoami, logout. Authentication flow -------------------- :: # Step 1: generate an Ed25519 key pair (private key stored in ~/.muse/keys/) muse auth keygen --hub https://musehub.ai # Step 2: register the public key with the hub via challenge-response muse auth register --hub https://musehub.ai --handle alice # Inspect stored identity: muse auth whoami Security model -------------- - ``_json_post`` validates the URL scheme (``http``/``https`` only) before making any network request — prevents SSRF from a tampered hub URL. - All diagnostic messages (progress, warnings, errors) go to **stderr**. **stdout** is reserved for machine-readable output and the interactive prompt (get-url returns a bare URL; --json returns a JSON object). Subcommands ----------- :: muse auth keygen [--hub HUB] [--label LABEL] [--force] [--json] muse auth register [--hub HUB] [--handle HANDLE] [--label LABEL] [--agent] [--json] muse auth whoami [--hub HUB] [--all] [--json] muse auth logout [--hub HUB] [--all] [--json] JSON schemas ------------ ``muse auth keygen --json``:: {"status": "ok", "hub": "", "hostname": "", "key_path": "", "public_key_b64": "", "fingerprint": ""} ``muse auth register --json``:: {"status": "registered"|"authenticated", "hub": "", "handle": "", "identity_type": "human"|"agent", "fingerprint": "", "identity_path": ""} ``muse auth whoami --json``:: {"hub": "", "type": "", "handle": "", "key_set": true, "capabilities": []} ``muse auth logout --json``:: {"status": "ok"|"nothing_to_do", "hubs": ["", ...], "count": } """ from __future__ import annotations import argparse import json import logging import os import pathlib import sys import urllib.error import urllib.parse import urllib.request from typing import TypedDict from muse.cli.config import get_hub_url from muse.core.errors import ExitCode from muse.core.identity import ( IdentityEntry, clear_all_identities, clear_identity, get_identity_path, hostname_from_url, list_all_identities, load_identity, save_identity, ) from muse.core.validation import sanitize_display logger = logging.getLogger(__name__) # Auth endpoints on the hub (relative to the hub base URL). _CHALLENGE_PATH = "/api/auth/challenge" _VERIFY_PATH = "/api/auth/verify" # Hard cap on response size to prevent OOM from a compromised hub. _MAX_RESPONSE_BYTES = 1 * 1024 * 1024 # 1 MiB # Only allow http and https — no file://, ftp://, data://, etc. _ALLOWED_SCHEMES = frozenset({"http", "https"}) # ── TypedDicts ──────────────────────────────────────────────────────────────── class _KeygenJson(TypedDict): """JSON schema for ``muse auth keygen --json``.""" status: str # "ok" hub: str # hub URL hostname: str # extracted hostname key_path: str # absolute path to private key PEM public_key_b64: str # base64url-encoded public key fingerprint: str # SHA-256 hex fingerprint class _RegisterJson(TypedDict): """JSON schema for ``muse auth register --json``.""" status: str # "registered" | "authenticated" hub: str # hub URL handle: str # registered username identity_id: str # hub-assigned ID identity_type: str # "human" | "agent" fingerprint: str # SHA-256 hex fingerprint of the public key token_stored: bool # always true on success identity_path: str # path to ~/.muse/identity.toml class _WhoamiJson(TypedDict): """JSON schema for ``muse auth whoami --json`` (per-identity entry).""" hub: str # hostname key type: str # "human" | "agent" | "" handle: str # registered handle fingerprint: str # SHA-256 hex of the public key key_set: bool # true if an Ed25519 key is stored capabilities: list[str] class _LogoutJson(TypedDict): """JSON schema for ``muse auth logout --json``.""" status: str # "ok" | "nothing_to_do" hubs: list[str] # hostnames logged out from count: int # number of identities removed class _JsonPayload(TypedDict, total=False): """Generic JSON request payload for hub auth endpoints (all fields optional str).""" fingerprint: str algorithm: str challenge_token: str public_key_b64: str signature_b64: str handle: str label: str class _ChallengeResp(TypedDict, total=False): """Parsed response from the hub challenge endpoint.""" challengeToken: str challenge_token: str isNewKey: bool is_new_key: bool class _VerifyResp(TypedDict, total=False): """Parsed response from the hub verify endpoint.""" handle: str identityId: str identity_id: str isNewIdentity: bool is_new_identity: bool # ── HTTP helpers ────────────────────────────────────────────────────────────── def _hub_base_url(hub_url: str) -> str: """Extract the base URL (scheme + host + port) from a full hub URL. Examples:: "https://musehub.ai/gabriel/muse" → "https://musehub.ai" "http://localhost:10003" → "http://localhost:10003" Raises: SystemExit: If the URL scheme is not ``http`` or ``https``. """ parsed = urllib.parse.urlparse(hub_url) if parsed.scheme.lower() not in _ALLOWED_SCHEMES: print( f"❌ Hub URL scheme '{sanitize_display(parsed.scheme)}' is not allowed. " f"Use http or https.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) port_str = f":{parsed.port}" if parsed.port else "" return f"{parsed.scheme}://{parsed.hostname}{port_str}" def _json_post_raw(base_url: str, path: str, payload: _JsonPayload) -> _JsonPayload: """POST *payload* as JSON and return the raw parsed response dict. Private implementation — call :func:`_post_challenge` or :func:`_post_verify` instead. Validates the URL scheme before any network I/O to prevent SSRF. Args: base_url: Hub base URL (e.g. ``"https://musehub.ai"``). path: Endpoint path (e.g. ``"/api/auth/challenge"``). payload: Dict to serialise as the JSON body (``None`` values omitted). Returns: Parsed JSON response body as a dict. Raises: SystemExit: On invalid URL scheme, HTTP error, or network failure. """ scheme = urllib.parse.urlparse(base_url).scheme.lower() if scheme not in _ALLOWED_SCHEMES: print( f"❌ Hub URL scheme '{sanitize_display(scheme)}' is not allowed. " f"Use http or https.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) url = f"{base_url.rstrip('/')}{path}" clean_payload = {k: v for k, v in payload.items() if v is not None} body_bytes = json.dumps(clean_payload).encode("utf-8") req = urllib.request.Request( url=url, data=body_bytes, headers={ "Content-Type": "application/json", "Accept": "application/json", }, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: # noqa: S310 raw: bytes = resp.read(_MAX_RESPONSE_BYTES + 1) except urllib.error.HTTPError as exc: try: err_body = exc.read().decode("utf-8", errors="replace")[:400] except Exception: # noqa: BLE001 err_body = "" print( f"❌ HTTP {exc.code} from {sanitize_display(url)}: {sanitize_display(err_body)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from exc except urllib.error.URLError as exc: print( f"❌ Network error contacting {sanitize_display(url)}: " f"{sanitize_display(str(exc.reason))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) from exc if len(raw) > _MAX_RESPONSE_BYTES: print(f"❌ Response from {sanitize_display(url)} exceeds size limit.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) parsed = json.loads(raw) if not isinstance(parsed, dict): print(f"❌ Unexpected response shape from {sanitize_display(url)}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return parsed def _post_challenge(base_url: str, payload: _JsonPayload) -> _ChallengeResp: """POST to the challenge endpoint and return a typed response.""" raw = _json_post_raw(base_url, _CHALLENGE_PATH, payload) return _ChallengeResp( challengeToken=str(raw.get("challengeToken") or ""), challenge_token=str(raw.get("challenge_token") or ""), isNewKey=bool(raw.get("isNewKey") or raw.get("is_new_key")), is_new_key=bool(raw.get("is_new_key")), ) def _post_verify(base_url: str, payload: _JsonPayload) -> _VerifyResp: """POST to the verify endpoint and return a typed response.""" raw = _json_post_raw(base_url, _VERIFY_PATH, payload) return _VerifyResp( handle=str(raw.get("handle") or ""), identityId=str(raw.get("identityId") or ""), identity_id=str(raw.get("identity_id") or ""), isNewIdentity=bool(raw.get("isNewIdentity") or raw.get("is_new_identity")), is_new_identity=bool(raw.get("is_new_identity")), ) # ── Helpers ─────────────────────────────────────────────────────────────────── def _resolve_hub(hub_opt: str | None, repo_root: pathlib.Path | None = None) -> str | None: """Return the hub URL: explicit option → repo config → None.""" if hub_opt: return hub_opt return get_hub_url(repo_root) def _display_entry(hostname: str, entry: IdentityEntry, *, json_output: bool) -> None: """Print an identity entry. JSON → stdout; human-readable → stderr.""" itype = entry.get("type") or "" handle = entry.get("handle") or "" fingerprint = entry.get("fingerprint") or "" key_path = entry.get("key_path") or "" caps = list(entry.get("capabilities") or []) if json_output: out: _WhoamiJson = { "hub": hostname, "type": itype, "handle": handle, "fingerprint": fingerprint, "key_set": bool(key_path), "capabilities": caps, } print(json.dumps(out)) else: print("", file=sys.stderr) print(f" Hub: {sanitize_display(hostname)}", file=sys.stderr) print(f" Type: {itype or 'unknown'}", file=sys.stderr) print(f" Handle: {sanitize_display(handle) or '—'}", file=sys.stderr) print(f" Fingerprint: {fingerprint[:16] + '…' if fingerprint else '—'}", file=sys.stderr) print(f" Key: {sanitize_display(key_path) or 'not set — run muse auth keygen'}", file=sys.stderr) if caps: print(f" Caps: {' '.join(caps)}", file=sys.stderr) print("", file=sys.stderr) # ── register ────────────────────────────────────────────────────────────────── def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``muse auth`` subcommand tree and all its flags. Every subcommand accepts ``--json`` for machine-readable output on stdout. All progress and diagnostic messages go to stderr. """ parser = subparsers.add_parser( "auth", help="Identity management.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) subs = parser.add_subparsers(dest="subcommand", metavar="SUBCOMMAND") subs.required = True # ── keygen ─────────────────────────────────────────────────────────────── keygen_p = subs.add_parser( "keygen", help="Generate a new Ed25519 keypair for public-key authentication.", description=( "Generates a fresh Ed25519 keypair and stores the private key at\n" "~/.muse/keys/{hostname}.pem (mode 0o600).\n" "The public key fingerprint is printed to stderr for verification.\n" "Run 'muse auth register' afterward to register the key with the hub." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) keygen_p.add_argument("--hub", default=None, metavar="URL", help="Hub URL (e.g. https://musehub.ai). Falls back to [hub] url in config.toml.") keygen_p.add_argument("--agent-id", default=None, metavar="AGENT_ID", dest="agent_id", help=( "Generate a dedicated keypair for this agent handle " "(stored at ~/.muse/keys/{hostname}__{agent_id}.pem). " "Omit for the human (operator) key." )) keygen_p.add_argument("--label", default=None, metavar="LABEL", help='Friendly key label, e.g. "MacBook Pro" or "CI agent". Stored locally.') keygen_p.add_argument("--force", action="store_true", help="Overwrite an existing key for this hub without prompting.") keygen_p.add_argument("--json", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.") keygen_p.set_defaults(func=run_keygen) # ── register ───────────────────────────────────────────────────────────── register_p = subs.add_parser( "register", help="Register the local Ed25519 key with a MuseHub instance (challenge-response).", description=( "Performs the Ed25519 challenge-response flow with the hub:\n" " 1. POST /api/auth/challenge (fingerprint → nonce)\n" " 2. Sign the nonce with the local private key\n" " 3. POST /api/auth/verify (signed nonce → session token)\n\n" "The resulting session token is saved to ~/.muse/identity.toml (0o600).\n" "Use this command for initial registration AND to refresh an expired token." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) register_p.add_argument("--hub", default=None, metavar="URL", help="Hub URL (e.g. https://musehub.ai). Falls back to [hub] url in config.toml.") register_p.add_argument("--handle", default=None, metavar="HANDLE", help=( "Desired username (required for first-time registration; " "ignored if the key is already registered)." )) register_p.add_argument("--label", default=None, metavar="LABEL", help='Friendly key label, e.g. "MacBook Pro" or "CI agent".') register_p.add_argument("--agent", action="store_true", help="Mark this identity as an agent (default: human).") register_p.add_argument("--agent-id", default=None, metavar="AGENT_ID", dest="agent_id", help=( "Agent handle whose dedicated key should be registered. " "Loads ~/.muse/keys/{hostname}__{agent_id}.pem and stores " "the identity under the compound key 'hostname#agent_id'. " "Implies --agent." )) register_p.add_argument("--provisioned-by", default=None, metavar="HANDLE", dest="provisioned_by", help=( "Handle of the human who is provisioning this agent key. " "Recorded in identity.toml as the trust-chain root. " "Required when --agent-id is used." )) register_p.add_argument("--json", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on success.") register_p.set_defaults(func=run_register) # ── whoami ──────────────────────────────────────────────────────────────── whoami_p = subs.add_parser( "whoami", help="Show the current identity stored in ~/.muse/identity.toml.", description=( "Print the identity stored in ~/.muse/identity.toml for a hub.\n" "Exits non-zero when no identity is stored — useful for agent branching:\n\n" " muse auth whoami --json || muse auth register --hub --handle --agent\n\n" "With --all --json, emits a single JSON array (one object per hub):\n\n" " muse auth whoami --all --json | jq '.[] | select(.type == \"agent\")'" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) whoami_p.add_argument("--hub", default=None, metavar="URL", help="Hub URL to inspect. Defaults to the repo's configured hub.") whoami_p.add_argument("--all", "-a", action="store_true", dest="all_hubs", help="Show identities for all configured hubs.") whoami_p.add_argument("--json", "-j", action="store_true", dest="json_output", help="Emit JSON to stdout. With --all, emits a JSON array.") whoami_p.set_defaults(func=run_whoami) # ── logout ──────────────────────────────────────────────────────────────── logout_p = subs.add_parser( "logout", help="Remove stored credentials for a hub.", description=( "Remove the signing identity stored in ~/.muse/identity.toml for a hub.\n" "Operation is idempotent — logging out when no identity is stored exits 0.\n\n" "Agent quickstart:\n" " muse auth logout --json # single hub, JSON result\n" " muse auth logout --all --json # remove all hubs, sorted list\n\n" "JSON output shape (stdout):\n" " {\"status\": \"ok\" | \"nothing_to_do\",\n" " \"hub\": \"\", # single-hub only\n" " \"hubs\": [\"\", ...], # --all only (sorted)\n" " \"count\": } # --all only\n\n" "Exit codes: 0 success (incl. nothing to do), 1 bad arguments, 3 internal error." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) logout_p.add_argument("--hub", default=None, metavar="URL", help="Hub URL to log out from. Defaults to the repo's configured hub.") logout_p.add_argument("--all", "-a", action="store_true", dest="all_hubs", help="Remove credentials for ALL configured hubs.") logout_p.add_argument("--json", "-j", action="store_true", dest="json_output", default=False, help="Emit a JSON object to stdout on completion.") logout_p.set_defaults(func=run_logout) # ── keygen ──────────────────────────────────────────────────────────────────── def run_keygen(args: argparse.Namespace) -> None: """Generate a new Ed25519 keypair for this hub (or for a specific agent). Human keys are written to ``~/.muse/keys/{hostname}.pem`` (0o600). Agent keys are written to ``~/.muse/keys/{hostname}__{agent_id}.pem``. All progress text goes to stderr. With ``--json``, a :class:`_KeygenJson` object is emitted to stdout so agents can capture the fingerprint and public key without parsing human-readable text. Run ``muse auth register`` immediately after to register the public key with the hub and obtain a session token. """ from muse.core.keypair import generate_keypair, key_path_for hub: str | None = args.hub agent_id: str | None = getattr(args, "agent_id", None) label: str | None = args.label force: bool = args.force json_output: bool = args.json_output hub_url = _resolve_hub(hub) if hub_url is None: print( "❌ No hub URL provided.\n" " Pass --hub , or first run: muse hub connect ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) hostname = hostname_from_url(hub_url) key_path = key_path_for(hostname, agent_id) if key_path.exists() and not force: subject = f"{sanitize_display(hostname)}#{sanitize_display(agent_id)}" if agent_id else sanitize_display(hostname) print( f"⚠️ A key already exists for {subject}: {key_path}\n" " Pass --force to overwrite it (you will need to re-register).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if key_path.exists() and force: subject = f"{sanitize_display(hostname)}#{sanitize_display(agent_id)}" if agent_id else sanitize_display(hostname) print(f"⚠️ Overwriting existing key for {subject}.", file=sys.stderr) pub_b64, fingerprint = generate_keypair(hostname, agent_id) if json_output: out: _KeygenJson = { "status": "ok", "hub": hub_url, "hostname": hostname, "key_path": str(key_path), "public_key_b64": pub_b64, "fingerprint": fingerprint, } print(json.dumps(out)) else: label_note = f" (label: {label!r})" if label else "" agent_note = f" (agent: {agent_id})" if agent_id else "" register_flags = f"--hub {hub_url} --handle " if agent_id: register_flags = f"--hub {hub_url} --agent-id {agent_id} --handle {agent_id} --provisioned-by " print( f"✅ Ed25519 keypair generated{label_note}{agent_note}\n" f" Private key: {key_path}\n" f" Public key (b64url): {pub_b64}\n" f" Fingerprint (SHA-256): {fingerprint}\n\n" f" Next step: muse auth register {register_flags}", file=sys.stderr, ) # ── register ────────────────────────────────────────────────────────────────── def run_register(args: argparse.Namespace) -> None: """Register the local Ed25519 key with a MuseHub instance. Performs the full challenge-response flow: 1. Derives the public key from the local private key. 2. Requests a challenge nonce from ``/api/auth/challenge``. 3. Signs the raw nonce bytes with the Ed25519 private key. 4. Submits the signature to ``/api/auth/verify``. 5. Stores the returned identity in ``~/.muse/identity.toml``. When ``--agent-id`` is supplied the agent's dedicated key (``~/.muse/keys/{hostname}__{agent_id}.pem``) is used and the entry is stored under the compound key ``"hostname#agent_id"``. All progress messages go to stderr. With ``--json``, a :class:`_RegisterJson` object is emitted to stdout on success — agents can capture the identity ID and token path without parsing text. Security notes: - The private key never leaves the local machine; only the signature is sent. - The nonce is signed as raw bytes — Ed25519 includes collision-resistant prehashing internally. - ``_json_post`` validates the hub URL scheme before any network I/O. """ from muse.core.keypair import ( key_path_for, load_private_key, public_key_fingerprint, public_key_to_b64url, sign_bytes, ) hub: str | None = args.hub handle: str | None = args.handle label: str | None = args.label agent: bool = args.agent agent_id: str | None = getattr(args, "agent_id", None) provisioned_by: str | None = getattr(args, "provisioned_by", None) json_output: bool = args.json_output # --agent-id implies --agent if agent_id: agent = True hub_url = _resolve_hub(hub) if hub_url is None: print( "❌ No hub URL provided.\n" " Pass --hub , or first run: muse hub connect ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if agent_id and not provisioned_by: print( "❌ --agent-id requires --provisioned-by .\n" " Example: muse auth register --agent-id agentception-abc123 " "--handle agentception-abc123 --provisioned-by gabriel", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) hostname = hostname_from_url(hub_url) base_url = _hub_base_url(hub_url) private_key = load_private_key(hostname, agent_id) if private_key is None: keygen_flags = f"--hub {hub_url}" if agent_id: keygen_flags += f" --agent-id {agent_id}" print( f"❌ No Ed25519 key found for {sanitize_display(hostname)}" + (f"#{sanitize_display(agent_id)}" if agent_id else "") + ".\n" f" Generate one first: muse auth keygen {keygen_flags}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) public_key = private_key.public_key() fingerprint = public_key_fingerprint(public_key) pub_b64 = public_key_to_b64url(public_key) # Step 1: request a challenge nonce. print(f" → Requesting challenge from {sanitize_display(base_url)} …", file=sys.stderr) challenge_resp = _post_challenge(base_url, { "fingerprint": fingerprint, "algorithm": "ed25519", }) challenge_token = challenge_resp.get("challengeToken") or challenge_resp.get("challenge_token") or "" if not challenge_token: print( "❌ Hub returned an invalid challenge response (missing challengeToken).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) is_new_key = challenge_resp.get("isNewKey") or challenge_resp.get("is_new_key") or False if is_new_key and not handle: print( "❌ This key is not yet registered. " "Pass --handle to register it.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) nonce_hex = challenge_token if not nonce_hex or not all(c in "0123456789abcdef" for c in nonce_hex): print(f"❌ Hub returned an invalid challenge (expected hex nonce).", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) try: nonce_bytes = bytes.fromhex(nonce_hex) except ValueError as exc: print(f"❌ Could not decode nonce: {sanitize_display(str(exc))}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc # Step 2: sign the nonce. signature_b64 = sign_bytes(private_key, nonce_bytes) # Step 3: submit the signed nonce. print(f" → Submitting signature to {sanitize_display(base_url)} …", file=sys.stderr) verify_resp = _post_verify(base_url, { "challenge_token": challenge_token, "public_key_b64": pub_b64, "signature_b64": signature_b64, "handle": handle, "label": label, }) returned_handle = verify_resp.get("handle") or handle or "" identity_id = verify_resp.get("identityId") or verify_resp.get("identity_id") or "" is_new_identity = verify_resp.get("isNewIdentity") or verify_resp.get("is_new_identity") or False if not returned_handle: print("❌ Hub returned an invalid verify response (missing handle).", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) identity_type = "agent" if agent else "human" # Store handle + key_path + fingerprint. # For agent keys, store under the compound key "hostname#agent_id" so that # human and agent entries coexist in identity.toml without collision. key_path = str(key_path_for(hostname, agent_id)) entry: IdentityEntry = { "type": identity_type, "handle": returned_handle, "key_path": key_path, "algorithm": "ed25519", "fingerprint": fingerprint, } if provisioned_by: entry["provisioned_by"] = provisioned_by try: save_identity(hub_url, entry, agent_id=agent_id) except OSError as exc: print(f"❌ Could not write credentials: {exc}", file=sys.stderr) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc action = "registered" if is_new_identity else "authenticated" identity_path = str(get_identity_path()) if json_output: result: _RegisterJson = { "status": action, "hub": hub_url, "handle": returned_handle, "identity_id": identity_id, "identity_type": identity_type, "fingerprint": fingerprint, "token_stored": False, "identity_path": identity_path, } print(json.dumps(result)) else: action_cap = action.capitalize() prov_line = f"\n Provisioned by: {provisioned_by}" if provisioned_by else "" print( f"\n✅ {action_cap} as '{returned_handle}' on {hub_url}{prov_line}\n" f" Identity ID: {identity_id}\n" f" Auth method: ed25519 (key fingerprint: {fingerprint[:16]}…)\n" f" Key path: {key_path}\n" f" Identity stored in: {identity_path}", file=sys.stderr, ) # ── whoami ──────────────────────────────────────────────────────────────────── def run_whoami(args: argparse.Namespace) -> None: """Show the identity stored in ``~/.muse/identity.toml`` for a hub. Exits non-zero when no identity is stored so agents can branch on authentication status:: muse auth whoami --hub http://localhost:10003 --json \\ || muse auth register --hub http://localhost:10003 --handle my-agent --agent JSON output (``--json``) ------------------------ Emits a :class:`_WhoamiJson` object to stdout:: { "hub": "", "type": "human" | "agent", "handle": "", "key_set": true | false, "capabilities": ["read", ...] ← only present when non-empty } With ``--all --json``, emits a **single JSON array** (one object per hub) so agents can pipe to ``jq``:: muse auth whoami --all --json | jq '.[] | select(.type == "agent")' Exit codes ---------- 0 Identity found and printed. 1 No hub configured, or no identity stored for that hub. """ hub: str | None = args.hub all_hubs: bool = args.all_hubs json_output: bool = args.json_output if all_hubs: identities = list_all_identities() if not identities: print("No identities stored. Run `muse auth keygen` + `muse auth register`.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if json_output: entries = [ { "hub": hostname, "type": e.get("type") or "", "handle": e.get("handle") or "", "fingerprint": e.get("fingerprint") or "", "key_set": bool(e.get("key_path")), "capabilities": list(e.get("capabilities") or []), } for hostname, e in sorted(identities.items()) ] print(json.dumps(entries)) else: for hostname, stored_entry in sorted(identities.items()): _display_entry(hostname, stored_entry, json_output=False) return hub_url = _resolve_hub(hub) if hub_url is None: # No hub in args or repo config — fall back to showing all identities. identities = list_all_identities() if not identities: print("No identities stored. Run `muse auth keygen` + `muse auth register`.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) if json_output: entries = [ { "hub": hostname, "type": e.get("type") or "", "handle": e.get("handle") or "", "fingerprint": e.get("fingerprint") or "", "key_set": bool(e.get("key_path")), "capabilities": list(e.get("capabilities") or []), } for hostname, e in sorted(identities.items()) ] print(json.dumps(entries)) else: for hostname, stored_entry in sorted(identities.items()): _display_entry(hostname, stored_entry, json_output=False) return single_entry = load_identity(hub_url) if single_entry is None: print( f"No identity stored for {sanitize_display(hub_url)}.\n" f"Run: muse auth keygen --hub {hub_url} && muse auth register --hub {hub_url} --handle ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) _display_entry(hostname_from_url(hub_url), single_entry, json_output=json_output) # ── logout ──────────────────────────────────────────────────────────────────── def run_logout(args: argparse.Namespace) -> None: """Remove stored credentials for one hub or all hubs. Deletes the matching entry (or entries) from ``~/.muse/identity.toml``. The hub URL in ``.muse/config.toml`` is **not** touched — use ``muse hub disconnect`` to remove the hub association from the repo too. Idempotent ---------- Calling ``logout`` when no identity is stored does **not** fail. The JSON response uses ``status: "nothing_to_do"`` so agents can distinguish "was logged in, now removed" from "was not logged in" without special-casing exit codes:: muse auth logout --hub http://localhost:10003 --json # → {"status": "nothing_to_do", "hubs": [], "count": 0} (exit 0) Agent quickstart ---------------- :: muse auth logout --all --json # clear every hub in one shot JSON output (``--json``) ------------------------ :: { "status": "ok" | "nothing_to_do", "hubs": ["hostname1", ...], ← sorted; empty on nothing_to_do "count": ← 0 on nothing_to_do } All diagnostic messages go to stderr regardless of ``--json``. Performance note ---------------- ``--all`` performs a single atomic read-modify-write via :func:`~muse.core.identity.clear_all_identities`, regardless of how many hubs are stored. It does **not** call :func:`~muse.core.identity.clear_identity` in a loop. Exit codes ---------- 0 Success (credentials removed or nothing to do). 1 No hub URL could be resolved (no ``--hub`` flag and no hub in config). """ hub: str | None = args.hub all_hubs: bool = args.all_hubs json_output: bool = args.json_output if all_hubs: removed_hubs = clear_all_identities() if not removed_hubs: if json_output: out: _LogoutJson = {"status": "nothing_to_do", "hubs": [], "count": 0} print(json.dumps(out)) else: print("No identities stored.", file=sys.stderr) return if json_output: result: _LogoutJson = { "status": "ok", "hubs": removed_hubs, # already sorted by clear_all_identities "count": len(removed_hubs), } print(json.dumps(result)) else: hub_list = ", ".join(sanitize_display(h) for h in removed_hubs) print( f"✅ Logged out from {len(removed_hubs)} hub(s): {hub_list}", file=sys.stderr, ) return hub_url = _resolve_hub(hub) if hub_url is None: print( "❌ No hub URL provided.\n" " Pass --hub , or first run: muse hub connect ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) removed = clear_identity(hub_url) hub_display = hostname_from_url(hub_url) if json_output: if removed: single_result: _LogoutJson = { "status": "ok", "hubs": [sanitize_display(hub_display)], "count": 1, } else: single_result = {"status": "nothing_to_do", "hubs": [], "count": 0} print(json.dumps(single_result)) else: if removed: print(f"✅ Logged out from {sanitize_display(hub_display)}.", file=sys.stderr) else: print( f"No identity stored for {sanitize_display(hub_display)} — nothing to do.", file=sys.stderr, )