gabriel / muse public
test_cmd_whoami_hardening.py python
334 lines 10.7 KB
Raw
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 150 days ago
1 """Hardening tests for ``muse whoami``.
2
3 Covers:
4 Unit — _build_entry_json (key_set bool, capabilities, handle/fingerprint)
5 _print_text (ANSI injection in hub/handle/type/fingerprint)
6 JSON — _WhoamiJson schema (key_set bool not string, all fields, capabilities)
7 Flags — -j/-a short flags, --all --json emits single parseable array
8 Integration — no hub configured → stderr + exit 1,
9 no identity stored → stderr + exit 1,
10 identity found (text and JSON),
11 --all lists multiple hubs
12 Stress — 50-hub --all --json
13 """
14
15 from __future__ import annotations
16
17 type _IdentityMap = dict[str, IdentityEntry]
18
19 import json
20 import pathlib
21 import threading
22 from contextlib import contextmanager
23 from typing import Generator, TypedDict
24 from unittest.mock import patch
25
26 from tests.cli_test_helper import CliRunner, InvokeResult
27 from muse.core.identity import IdentityEntry
28
29 cli = None
30 runner = CliRunner()
31 _invoke_lock = threading.Lock()
32
33 _HUB = "localhost:10003"
34 _HUB_URL = f"http://{_HUB}"
35
36
37 class _WhoamiOut(TypedDict, total=False):
38 hub: str
39 type: str
40 handle: str
41 fingerprint: str
42 key_set: bool
43 capabilities: list[str]
44
45
46 def _invoke(args: list[str]) -> InvokeResult:
47 with _invoke_lock:
48 return runner.invoke(cli, args)
49
50
51 def _make_identity(
52 *,
53 itype: str = "human",
54 handle: str = "alice",
55 fingerprint: str = "fp123abc",
56 key_path: str = "/home/alice/.muse/keys/hub.pem",
57 capabilities: list[str] | None = None,
58 ) -> IdentityEntry:
59 entry: IdentityEntry = {
60 "type": itype,
61 "handle": handle,
62 "fingerprint": fingerprint,
63 "key_path": key_path,
64 "algorithm": "ed25519",
65 }
66 if capabilities is not None:
67 entry["capabilities"] = capabilities
68 return entry
69
70
71 @contextmanager
72 def _patch_identity(
73 hub_url: str | None = _HUB_URL,
74 identity: IdentityEntry | None = None,
75 all_identities: _IdentityMap | None = None,
76 ) -> Generator[None, None, None]:
77 with patch("muse.cli.commands.whoami.get_hub_url", return_value=hub_url), \
78 patch("muse.cli.commands.whoami.load_identity", return_value=identity), \
79 patch("muse.cli.commands.whoami.list_all_identities", return_value=all_identities or {}):
80 yield
81
82
83 # ---------------------------------------------------------------------------
84 # Unit: _build_entry_json
85 # ---------------------------------------------------------------------------
86
87
88 def test_build_entry_json_key_set_is_bool_true() -> None:
89 from muse.cli.commands.whoami import _build_entry_json
90 entry = _make_identity()
91 out = _build_entry_json(_HUB, entry)
92 assert out["key_set"] is True
93 assert isinstance(out["key_set"], bool)
94
95
96 def test_build_entry_json_key_set_is_bool_false() -> None:
97 from muse.cli.commands.whoami import _build_entry_json
98 entry: IdentityEntry = {"type": "human", "handle": "alice"}
99 out = _build_entry_json(_HUB, entry)
100 assert out["key_set"] is False
101 assert isinstance(out["key_set"], bool)
102
103
104 def test_build_entry_json_capabilities_included() -> None:
105 from muse.cli.commands.whoami import _build_entry_json
106 entry = _make_identity(capabilities=["read:*", "write:midi"])
107 out = _build_entry_json(_HUB, entry)
108 assert out.get("capabilities") == ["read:*", "write:midi"]
109
110
111 def test_build_entry_json_empty_capabilities_omitted() -> None:
112 from muse.cli.commands.whoami import _build_entry_json
113 entry = _make_identity(capabilities=[])
114 out = _build_entry_json(_HUB, entry)
115 assert out.get("capabilities") is None or out.get("capabilities") == []
116
117
118 def test_build_entry_json_handle_and_fingerprint() -> None:
119 from muse.cli.commands.whoami import _build_entry_json
120 entry = _make_identity(handle="gabriel", fingerprint="deadbeef")
121 out = _build_entry_json(_HUB, entry)
122 assert out["handle"] == "gabriel"
123 assert out["fingerprint"] == "deadbeef"
124
125
126 # ---------------------------------------------------------------------------
127 # Unit: _print_text — ANSI injection
128 # ---------------------------------------------------------------------------
129
130
131 def test_ansi_injection_in_handle(tmp_path: pathlib.Path) -> None:
132 from muse.cli.commands.whoami import _print_text
133 import io, sys
134 evil_handle = "\x1b[31mevil\x1b[0m"
135 entry = _make_identity(handle=evil_handle)
136 buf = io.StringIO()
137 old = sys.stdout; sys.stdout = buf
138 try:
139 _print_text(_HUB, entry)
140 finally:
141 sys.stdout = old
142 assert "\x1b[" not in buf.getvalue()
143
144
145 def test_ansi_injection_in_hub(tmp_path: pathlib.Path) -> None:
146 from muse.cli.commands.whoami import _print_text
147 import io, sys
148 evil_hub = "\x1b[31mevil-hub\x1b[0m"
149 entry = _make_identity()
150 buf = io.StringIO()
151 old = sys.stdout; sys.stdout = buf
152 try:
153 _print_text(evil_hub, entry)
154 finally:
155 sys.stdout = old
156 assert "\x1b[" not in buf.getvalue()
157
158
159 # ---------------------------------------------------------------------------
160 # JSON schema
161 # ---------------------------------------------------------------------------
162
163
164 def test_json_schema_all_fields() -> None:
165 identity = _make_identity()
166 with _patch_identity(identity=identity):
167 result = _invoke(["whoami", "--json"])
168 assert result.exit_code == 0
169 data = json.loads(result.output)
170 for key in ("hub", "type", "handle", "fingerprint", "key_set"):
171 assert key in data, f"Missing key: {key}"
172
173
174 def test_json_key_set_false_when_no_key() -> None:
175 identity: IdentityEntry = {"type": "human", "handle": "alice"}
176 with _patch_identity(identity=identity):
177 result = _invoke(["whoami", "--json"])
178 assert result.exit_code == 0
179 data = json.loads(result.output)
180 assert data["key_set"] is False
181
182
183 def test_json_key_set_is_not_string() -> None:
184 identity = _make_identity()
185 with _patch_identity(identity=identity):
186 result = _invoke(["whoami", "--json"])
187 assert result.exit_code == 0
188 raw = result.output
189 assert '"key_set": true' in raw or '"key_set":true' in raw
190 assert '"key_set": "true"' not in raw
191
192
193 def test_json_capabilities_list() -> None:
194 identity = _make_identity(capabilities=["push", "pull"])
195 with _patch_identity(identity=identity):
196 result = _invoke(["whoami", "--json"])
197 assert result.exit_code == 0
198 data = json.loads(result.output)
199 assert data.get("capabilities") == ["push", "pull"]
200
201
202 # ---------------------------------------------------------------------------
203 # Flags
204 # ---------------------------------------------------------------------------
205
206
207 def test_short_j_flag_is_json() -> None:
208 identity = _make_identity()
209 with _patch_identity(identity=identity):
210 result = _invoke(["whoami", "-j"])
211 assert result.exit_code == 0
212 json.loads(result.output)
213
214
215 def test_short_a_flag_is_all() -> None:
216 identities = {
217 "hub-a.example.com": _make_identity(handle="alice"),
218 "hub-b.example.com": _make_identity(itype="agent", handle="bot"),
219 }
220 with _patch_identity(all_identities=identities):
221 result = _invoke(["whoami", "-a"])
222 assert result.exit_code == 0
223 assert "hub-a.example.com" in result.output or "hub-b.example.com" in result.output
224
225
226 def test_all_json_emits_array() -> None:
227 identities = {
228 "hub-a.example.com": _make_identity(handle="alice"),
229 "hub-b.example.com": _make_identity(itype="agent", handle="bot"),
230 }
231 with _patch_identity(all_identities=identities):
232 result = _invoke(["whoami", "--all", "--json"])
233 assert result.exit_code == 0
234 parsed = json.loads(result.output)
235 assert isinstance(parsed, list)
236 assert len(parsed) == 2
237
238
239 def test_all_json_each_entry_has_key_set_bool() -> None:
240 identities = {
241 f"hub{i}.example.com": _make_identity(handle=f"user{i}")
242 for i in range(3)
243 }
244 with _patch_identity(all_identities=identities):
245 result = _invoke(["whoami", "--all", "--json"])
246 assert result.exit_code == 0
247 parsed = json.loads(result.output)
248 for entry in parsed:
249 assert isinstance(entry["key_set"], bool)
250
251
252 def test_all_json_is_single_json_value() -> None:
253 identities = {f"hub{i}.example.com": _make_identity() for i in range(3)}
254 with _patch_identity(all_identities=identities):
255 result = _invoke(["whoami", "--all", "--json"])
256 assert result.exit_code == 0
257 json.loads(result.output) # must be a single JSON value
258
259
260 # ---------------------------------------------------------------------------
261 # Integration
262 # ---------------------------------------------------------------------------
263
264
265 def test_no_hub_configured_exits_1() -> None:
266 with _patch_identity(hub_url=None):
267 result = _invoke(["whoami"])
268 assert result.exit_code != 0
269
270
271 def test_no_identity_stored_exits_1() -> None:
272 with _patch_identity(identity=None):
273 result = _invoke(["whoami"])
274 assert result.exit_code != 0
275
276
277 def test_no_identities_for_all_exits_1() -> None:
278 with _patch_identity(all_identities={}):
279 result = _invoke(["whoami", "--all"])
280 assert result.exit_code != 0
281
282
283 def test_text_shows_handle_and_type() -> None:
284 identity = _make_identity(itype="agent", handle="worker")
285 with _patch_identity(identity=identity):
286 result = _invoke(["whoami"])
287 assert result.exit_code == 0
288 assert "worker" in result.output
289 assert "agent" in result.output
290
291
292 def test_text_all_shows_multiple_hubs() -> None:
293 identities = {
294 "hub-a.example.com": _make_identity(handle="alice"),
295 "hub-b.example.com": _make_identity(itype="agent", handle="bot"),
296 }
297 with _patch_identity(all_identities=identities):
298 result = _invoke(["whoami", "--all"])
299 assert result.exit_code == 0
300 assert "hub-a.example.com" in result.output
301 assert "hub-b.example.com" in result.output
302
303
304 def test_help_mentions_json() -> None:
305 result = _invoke(["whoami", "--help"])
306 assert "json" in result.output.lower()
307
308
309 def test_help_mentions_all() -> None:
310 result = _invoke(["whoami", "--help"])
311 assert "all" in result.output.lower()
312
313
314 # ---------------------------------------------------------------------------
315 # Stress
316 # ---------------------------------------------------------------------------
317
318
319 def test_stress_all_json_50_hubs() -> None:
320 identities = {
321 f"hub-{i:02d}.example.com": _make_identity(
322 handle=f"user-{i:02d}",
323 itype="agent" if i % 2 == 0 else "human",
324 )
325 for i in range(50)
326 }
327 with _patch_identity(all_identities=identities):
328 result = _invoke(["whoami", "--all", "--json"])
329 assert result.exit_code == 0
330 parsed = json.loads(result.output)
331 assert isinstance(parsed, list)
332 assert len(parsed) == 50
333 for entry in parsed:
334 assert isinstance(entry["key_set"], bool)
File History 1 commit
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 150 days ago