"""Hardening tests for ``muse whoami``. Covers: Unit — _build_entry_json (key_set bool, capabilities, handle/fingerprint) _print_text (ANSI injection in hub/handle/type/fingerprint) JSON — _WhoamiJson schema (key_set bool not string, all fields, capabilities) Flags — -j/-a short flags, --all --json emits single parseable array Integration — no hub configured → stderr + exit 1, no identity stored → stderr + exit 1, identity found (text and JSON), --all lists multiple hubs Stress — 50-hub --all --json """ from __future__ import annotations type _IdentityMap = dict[str, IdentityEntry] import json import pathlib import threading from contextlib import contextmanager from typing import Generator, TypedDict from unittest.mock import patch from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.identity import IdentityEntry cli = None runner = CliRunner() _invoke_lock = threading.Lock() _HUB = "localhost:10003" _HUB_URL = f"http://{_HUB}" class _WhoamiOut(TypedDict, total=False): hub: str type: str handle: str fingerprint: str key_set: bool capabilities: list[str] def _invoke(args: list[str]) -> InvokeResult: with _invoke_lock: return runner.invoke(cli, args) def _make_identity( *, itype: str = "human", handle: str = "alice", fingerprint: str = "fp123abc", key_path: str = "/home/alice/.muse/keys/hub.pem", capabilities: list[str] | None = None, ) -> IdentityEntry: entry: IdentityEntry = { "type": itype, "handle": handle, "fingerprint": fingerprint, "key_path": key_path, "algorithm": "ed25519", } if capabilities is not None: entry["capabilities"] = capabilities return entry @contextmanager def _patch_identity( hub_url: str | None = _HUB_URL, identity: IdentityEntry | None = None, all_identities: _IdentityMap | None = None, ) -> Generator[None, None, None]: with patch("muse.cli.commands.whoami.get_hub_url", return_value=hub_url), \ patch("muse.cli.commands.whoami.load_identity", return_value=identity), \ patch("muse.cli.commands.whoami.list_all_identities", return_value=all_identities or {}): yield # --------------------------------------------------------------------------- # Unit: _build_entry_json # --------------------------------------------------------------------------- def test_build_entry_json_key_set_is_bool_true() -> None: from muse.cli.commands.whoami import _build_entry_json entry = _make_identity() out = _build_entry_json(_HUB, entry) assert out["key_set"] is True assert isinstance(out["key_set"], bool) def test_build_entry_json_key_set_is_bool_false() -> None: from muse.cli.commands.whoami import _build_entry_json entry: IdentityEntry = {"type": "human", "handle": "alice"} out = _build_entry_json(_HUB, entry) assert out["key_set"] is False assert isinstance(out["key_set"], bool) def test_build_entry_json_capabilities_included() -> None: from muse.cli.commands.whoami import _build_entry_json entry = _make_identity(capabilities=["read:*", "write:midi"]) out = _build_entry_json(_HUB, entry) assert out.get("capabilities") == ["read:*", "write:midi"] def test_build_entry_json_empty_capabilities_omitted() -> None: from muse.cli.commands.whoami import _build_entry_json entry = _make_identity(capabilities=[]) out = _build_entry_json(_HUB, entry) assert out.get("capabilities") is None or out.get("capabilities") == [] def test_build_entry_json_handle_and_fingerprint() -> None: from muse.cli.commands.whoami import _build_entry_json entry = _make_identity(handle="gabriel", fingerprint="deadbeef") out = _build_entry_json(_HUB, entry) assert out["handle"] == "gabriel" assert out["fingerprint"] == "deadbeef" # --------------------------------------------------------------------------- # Unit: _print_text — ANSI injection # --------------------------------------------------------------------------- def test_ansi_injection_in_handle(tmp_path: pathlib.Path) -> None: from muse.cli.commands.whoami import _print_text import io, sys evil_handle = "\x1b[31mevil\x1b[0m" entry = _make_identity(handle=evil_handle) buf = io.StringIO() old = sys.stdout; sys.stdout = buf try: _print_text(_HUB, entry) finally: sys.stdout = old assert "\x1b[" not in buf.getvalue() def test_ansi_injection_in_hub(tmp_path: pathlib.Path) -> None: from muse.cli.commands.whoami import _print_text import io, sys evil_hub = "\x1b[31mevil-hub\x1b[0m" entry = _make_identity() buf = io.StringIO() old = sys.stdout; sys.stdout = buf try: _print_text(evil_hub, entry) finally: sys.stdout = old assert "\x1b[" not in buf.getvalue() # --------------------------------------------------------------------------- # JSON schema # --------------------------------------------------------------------------- def test_json_schema_all_fields() -> None: identity = _make_identity() with _patch_identity(identity=identity): result = _invoke(["whoami", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) for key in ("hub", "type", "handle", "fingerprint", "key_set"): assert key in data, f"Missing key: {key}" def test_json_key_set_false_when_no_key() -> None: identity: IdentityEntry = {"type": "human", "handle": "alice"} with _patch_identity(identity=identity): result = _invoke(["whoami", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data["key_set"] is False def test_json_key_set_is_not_string() -> None: identity = _make_identity() with _patch_identity(identity=identity): result = _invoke(["whoami", "--json"]) assert result.exit_code == 0 raw = result.output assert '"key_set": true' in raw or '"key_set":true' in raw assert '"key_set": "true"' not in raw def test_json_capabilities_list() -> None: identity = _make_identity(capabilities=["push", "pull"]) with _patch_identity(identity=identity): result = _invoke(["whoami", "--json"]) assert result.exit_code == 0 data = json.loads(result.output) assert data.get("capabilities") == ["push", "pull"] # --------------------------------------------------------------------------- # Flags # --------------------------------------------------------------------------- def test_short_j_flag_is_json() -> None: identity = _make_identity() with _patch_identity(identity=identity): result = _invoke(["whoami", "-j"]) assert result.exit_code == 0 json.loads(result.output) def test_short_a_flag_is_all() -> None: identities = { "hub-a.example.com": _make_identity(handle="alice"), "hub-b.example.com": _make_identity(itype="agent", handle="bot"), } with _patch_identity(all_identities=identities): result = _invoke(["whoami", "-a"]) assert result.exit_code == 0 assert "hub-a.example.com" in result.output or "hub-b.example.com" in result.output def test_all_json_emits_array() -> None: identities = { "hub-a.example.com": _make_identity(handle="alice"), "hub-b.example.com": _make_identity(itype="agent", handle="bot"), } with _patch_identity(all_identities=identities): result = _invoke(["whoami", "--all", "--json"]) assert result.exit_code == 0 parsed = json.loads(result.output) assert isinstance(parsed, list) assert len(parsed) == 2 def test_all_json_each_entry_has_key_set_bool() -> None: identities = { f"hub{i}.example.com": _make_identity(handle=f"user{i}") for i in range(3) } with _patch_identity(all_identities=identities): result = _invoke(["whoami", "--all", "--json"]) assert result.exit_code == 0 parsed = json.loads(result.output) for entry in parsed: assert isinstance(entry["key_set"], bool) def test_all_json_is_single_json_value() -> None: identities = {f"hub{i}.example.com": _make_identity() for i in range(3)} with _patch_identity(all_identities=identities): result = _invoke(["whoami", "--all", "--json"]) assert result.exit_code == 0 json.loads(result.output) # must be a single JSON value # --------------------------------------------------------------------------- # Integration # --------------------------------------------------------------------------- def test_no_hub_configured_exits_1() -> None: with _patch_identity(hub_url=None): result = _invoke(["whoami"]) assert result.exit_code != 0 def test_no_identity_stored_exits_1() -> None: with _patch_identity(identity=None): result = _invoke(["whoami"]) assert result.exit_code != 0 def test_no_identities_for_all_exits_1() -> None: with _patch_identity(all_identities={}): result = _invoke(["whoami", "--all"]) assert result.exit_code != 0 def test_text_shows_handle_and_type() -> None: identity = _make_identity(itype="agent", handle="worker") with _patch_identity(identity=identity): result = _invoke(["whoami"]) assert result.exit_code == 0 assert "worker" in result.output assert "agent" in result.output def test_text_all_shows_multiple_hubs() -> None: identities = { "hub-a.example.com": _make_identity(handle="alice"), "hub-b.example.com": _make_identity(itype="agent", handle="bot"), } with _patch_identity(all_identities=identities): result = _invoke(["whoami", "--all"]) assert result.exit_code == 0 assert "hub-a.example.com" in result.output assert "hub-b.example.com" in result.output def test_help_mentions_json() -> None: result = _invoke(["whoami", "--help"]) assert "json" in result.output.lower() def test_help_mentions_all() -> None: result = _invoke(["whoami", "--help"]) assert "all" in result.output.lower() # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- def test_stress_all_json_50_hubs() -> None: identities = { f"hub-{i:02d}.example.com": _make_identity( handle=f"user-{i:02d}", itype="agent" if i % 2 == 0 else "human", ) for i in range(50) } with _patch_identity(all_identities=identities): result = _invoke(["whoami", "--all", "--json"]) assert result.exit_code == 0 parsed = json.loads(result.output) assert isinstance(parsed, list) assert len(parsed) == 50 for entry in parsed: assert isinstance(entry["key_set"], bool)