gabriel / muse public
test_cmd_sign.py python
542 lines 21.5 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago
1 """Tests for ``muse sign`` CLI subcommands.
2
3 Coverage:
4 - run_header: text output, JSON output, path+hub vs full URL
5 - run_verify: valid → exit 0, invalid sig → exit 1, expired → exit 1, JSON output
6 - run_whoami: env-var source, identity.toml source, JSON output
7 - run_curl: correct curl command format, Authorization header embedded
8 - run_payment: JSON output fields, domain separation from HTTP MSign
9 - _load_signing: exit 1 on missing identity, key-path override
10 """
11
12 from __future__ import annotations
13
14 import argparse
15 import base64
16 import io
17 import json
18 import sys
19 import time
20 import unittest
21 import unittest.mock
22 from unittest.mock import MagicMock, patch
23
24
25 # ---------------------------------------------------------------------------
26 # Shared helpers
27 # ---------------------------------------------------------------------------
28
29 def _make_signing(handle: str = "gabriel") -> object:
30 """Return a real SigningIdentity with a fresh Ed25519 key."""
31 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
32 from muse.core.transport import SigningIdentity
33
34 return SigningIdentity(handle=handle, private_key=Ed25519PrivateKey.generate())
35
36
37 def _public_key_b64(signing: object) -> str:
38 """Return URL-safe base64 public key (no padding) for a SigningIdentity."""
39 pub_raw = signing.private_key.public_key().public_bytes_raw() # type: ignore[attr-defined]
40 return base64.urlsafe_b64encode(pub_raw).rstrip(b"=").decode("ascii")
41
42
43 def _make_header(signing: object, method: str, url: str,
44 body: bytes = b"", ts: int = 1744000000) -> str:
45 """Build an MSign header value for testing verify subcommand."""
46 from muse.core.msign import build_msign_header
47 return build_msign_header(signing, method, url, body, ts=ts)
48
49
50 def _run_cmd(func, **kwargs) -> argparse.Namespace:
51 """Build an argparse.Namespace from kwargs and call func(args)."""
52 args = argparse.Namespace(**kwargs)
53 func(args)
54 return args
55
56
57 # ---------------------------------------------------------------------------
58 # run_header
59 # ---------------------------------------------------------------------------
60
61 class TestRunHeader(unittest.TestCase):
62 def setUp(self) -> None:
63 self.signing = _make_signing("gabriel")
64
65 def _args(self, **kwargs) -> argparse.Namespace:
66 defaults = dict(
67 method="POST",
68 path=None,
69 url="https://hub.example.com/gabriel/muse/push",
70 hub=None,
71 body=None,
72 body_file=None,
73 timestamp=1744000000,
74 key_path=None,
75 agent_id=None,
76 json=False,
77 format="text",
78 )
79 defaults.update(kwargs)
80 return argparse.Namespace(**defaults)
81
82 def test_text_output_is_msign_header(self) -> None:
83 from muse.cli.commands.sign import run_header
84
85 args = self._args()
86 with patch("muse.cli.commands.sign._load_signing", return_value=self.signing), \
87 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
88 run_header(args)
89 output = mock_out.getvalue().strip()
90 assert output.startswith('MSign handle="gabriel"'), f"Unexpected: {output!r}"
91 assert "ts=1744000000" in output
92 assert ' sig="' in output
93
94 def test_json_output_fields(self) -> None:
95 from muse.cli.commands.sign import run_header
96
97 args = self._args(json=True)
98 with patch("muse.cli.commands.sign._load_signing", return_value=self.signing), \
99 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
100 run_header(args)
101 data = json.loads(mock_out.getvalue())
102 assert data["handle"] == "gabriel"
103 assert data["method"] == "POST"
104 assert data["timestamp"] == 1744000000
105 assert data["algorithm"] == "ed25519"
106 assert "signature_b64" in data
107 assert "fingerprint" in data
108 assert "body_sha256" in data
109 assert data["header_value"].startswith("MSign")
110
111 def test_path_plus_hub_constructs_url(self) -> None:
112 """--path + --hub must construct a full URL for signing."""
113 from muse.cli.commands.sign import run_header
114
115 args = self._args(
116 path="/gabriel/muse/push",
117 url=None,
118 hub="https://staging.musehub.ai",
119 json=True,
120 )
121 with patch("muse.cli.commands.sign._load_signing", return_value=self.signing), \
122 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
123 run_header(args)
124 data = json.loads(mock_out.getvalue())
125 assert data["path"] == "/gabriel/muse/push"
126
127 def test_empty_body_uses_empty_sha256(self) -> None:
128 """No body → body_sha256 == sha256(b'')."""
129 import hashlib
130 from muse.cli.commands.sign import run_header
131
132 args = self._args(json=True)
133 with patch("muse.cli.commands.sign._load_signing", return_value=self.signing), \
134 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
135 run_header(args)
136 data = json.loads(mock_out.getvalue())
137 expected = hashlib.sha256(b"").hexdigest()
138 assert data["body_sha256"] == expected
139
140 def test_inline_body_reflected_in_sha256(self) -> None:
141 """--body flag must be hashed into body_sha256."""
142 import hashlib
143 from muse.cli.commands.sign import run_header
144
145 args = self._args(body="hello world", json=True)
146 with patch("muse.cli.commands.sign._load_signing", return_value=self.signing), \
147 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
148 run_header(args)
149 data = json.loads(mock_out.getvalue())
150 expected = hashlib.sha256(b"hello world").hexdigest()
151 assert data["body_sha256"] == expected
152
153
154 # ---------------------------------------------------------------------------
155 # run_verify
156 # ---------------------------------------------------------------------------
157
158 class TestRunVerify(unittest.TestCase):
159 def setUp(self) -> None:
160 self.signing = _make_signing("gabriel")
161 self.pub_b64 = _public_key_b64(self.signing)
162 self.url = "https://hub.example.com/gabriel/muse/push"
163 self.ts = int(time.time())
164 self.header = _make_header(self.signing, "POST", self.url, b"", self.ts)
165
166 def _args(self, **kwargs) -> argparse.Namespace:
167 defaults = dict(
168 header=self.header,
169 method="POST",
170 url=self.url,
171 public_key_b64=self.pub_b64,
172 body=None,
173 body_file=None,
174 max_age=30,
175 json=False,
176 format="text",
177 )
178 defaults.update(kwargs)
179 return argparse.Namespace(**defaults)
180
181 def test_valid_header_exits_0(self) -> None:
182 from muse.cli.commands.sign import run_verify
183
184 args = self._args()
185 # Should not raise SystemExit.
186 run_verify(args)
187
188 def test_valid_header_json_output(self) -> None:
189 from muse.cli.commands.sign import run_verify
190
191 args = self._args(json=True)
192 with patch("sys.stdout", new_callable=io.StringIO) as mock_out:
193 run_verify(args)
194 data = json.loads(mock_out.getvalue())
195 assert data["valid"] is True
196 assert data["reason"] == "ok"
197
198 def test_wrong_public_key_exits_1(self) -> None:
199 from muse.cli.commands.sign import run_verify
200
201 other = _make_signing("other")
202 wrong_pub = _public_key_b64(other)
203 args = self._args(public_key_b64=wrong_pub)
204 with self.assertRaises(SystemExit) as cm:
205 run_verify(args)
206 assert cm.exception.code == 1
207
208 def test_wrong_public_key_json_valid_false(self) -> None:
209 from muse.cli.commands.sign import run_verify
210
211 other = _make_signing("other")
212 wrong_pub = _public_key_b64(other)
213 args = self._args(public_key_b64=wrong_pub, json=True)
214 with patch("sys.stdout", new_callable=io.StringIO) as mock_out:
215 with self.assertRaises(SystemExit):
216 run_verify(args)
217 data = json.loads(mock_out.getvalue())
218 assert data["valid"] is False
219 assert "reason" in data
220
221 def test_expired_timestamp_exits_1(self) -> None:
222 from muse.cli.commands.sign import run_verify
223
224 old_ts = int(time.time()) - 9999
225 old_header = _make_header(self.signing, "POST", self.url, b"", old_ts)
226 args = self._args(header=old_header, max_age=30)
227 with self.assertRaises(SystemExit) as cm:
228 run_verify(args)
229 assert cm.exception.code == 1
230
231 def test_malformed_header_exits_1(self) -> None:
232 from muse.cli.commands.sign import run_verify
233
234 args = self._args(header="Bearer garbage", json=True)
235 with patch("sys.stdout", new_callable=io.StringIO) as mock_out:
236 with self.assertRaises(SystemExit) as cm:
237 run_verify(args)
238 assert cm.exception.code == 1
239 data = json.loads(mock_out.getvalue())
240 assert data["valid"] is False
241
242 def test_method_mismatch_exits_1(self) -> None:
243 """A header signed for POST must not verify for GET."""
244 from muse.cli.commands.sign import run_verify
245
246 args = self._args(method="GET")
247 with self.assertRaises(SystemExit) as cm:
248 run_verify(args)
249 assert cm.exception.code == 1
250
251 def test_body_mismatch_exits_1(self) -> None:
252 """Changing the body after signing must invalidate the header."""
253 from muse.cli.commands.sign import run_verify
254
255 args = self._args(body="tampered")
256 with self.assertRaises(SystemExit) as cm:
257 run_verify(args)
258 assert cm.exception.code == 1
259
260 def test_custom_max_age_accepted(self) -> None:
261 """A very old timestamp is accepted if max_age is large enough."""
262 from muse.cli.commands.sign import run_verify
263
264 old_ts = int(time.time()) - 9999
265 old_header = _make_header(self.signing, "POST", self.url, b"", old_ts)
266 args = self._args(header=old_header, max_age=99999)
267 # Should not raise.
268 run_verify(args)
269
270
271 # ---------------------------------------------------------------------------
272 # run_curl
273 # ---------------------------------------------------------------------------
274
275 class TestRunCurl(unittest.TestCase):
276 def _args(self, **kwargs) -> argparse.Namespace:
277 defaults = dict(
278 method="POST",
279 url="https://hub.example.com/gabriel/muse/push",
280 hub=None,
281 body=None,
282 body_file=None,
283 content_type="application/json",
284 timestamp=1744000000,
285 key_path=None,
286 agent_id=None,
287 json=False,
288 format="text",
289 )
290 defaults.update(kwargs)
291 return argparse.Namespace(**defaults)
292
293 def test_curl_starts_with_curl(self) -> None:
294 from muse.cli.commands.sign import run_curl
295
296 signing = _make_signing()
297 args = self._args()
298 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
299 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
300 run_curl(args)
301 output = mock_out.getvalue()
302 assert output.strip().startswith("curl -X POST")
303
304 def test_authorization_header_embedded(self) -> None:
305 """The curl command must include the MSign Authorization header."""
306 from muse.cli.commands.sign import run_curl
307
308 signing = _make_signing()
309 args = self._args()
310 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
311 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
312 run_curl(args)
313 output = mock_out.getvalue()
314 assert "Authorization: MSign" in output
315
316 def test_url_appears_in_output(self) -> None:
317 from muse.cli.commands.sign import run_curl
318
319 signing = _make_signing()
320 url = "https://hub.example.com/gabriel/muse/push"
321 args = self._args(url=url)
322 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
323 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
324 run_curl(args)
325 assert url in mock_out.getvalue()
326
327 def test_body_file_uses_data_binary(self) -> None:
328 """--body-file must produce --data-binary @filename."""
329 from muse.cli.commands.sign import run_curl
330
331 signing = _make_signing()
332 args = self._args(body_file="/tmp/payload.bin")
333 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
334 patch("muse.cli.commands.sign._read_body", return_value=b"payload"), \
335 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
336 run_curl(args)
337 output = mock_out.getvalue()
338 assert "--data-binary @/tmp/payload.bin" in output
339
340 def test_inline_body_uses_data_flag(self) -> None:
341 """--body STRING must produce --data '...'."""
342 from muse.cli.commands.sign import run_curl
343
344 signing = _make_signing()
345 args = self._args(body='{"key": "value"}')
346 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
347 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
348 run_curl(args)
349 output = mock_out.getvalue()
350 assert "--data" in output
351 assert '{"key": "value"}' in output
352
353 def test_get_method(self) -> None:
354 from muse.cli.commands.sign import run_curl
355
356 signing = _make_signing()
357 args = self._args(method="GET")
358 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
359 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
360 run_curl(args)
361 assert "curl -X GET" in mock_out.getvalue()
362
363
364 # ---------------------------------------------------------------------------
365 # run_payment
366 # ---------------------------------------------------------------------------
367
368 class TestRunPayment(unittest.TestCase):
369 _NONCE = "a" * 64 # 64-char hex nonce
370
371 def _args(self, **kwargs) -> argparse.Namespace:
372 defaults = dict(
373 from_handle="alice",
374 to_handle="bob",
375 amount=1_000_000,
376 nonce=self._NONCE,
377 currency="nanoMUSE",
378 memo="stem:sha256:abc123",
379 hub=None,
380 key_path=None,
381 agent_id=None,
382 timestamp=1744000000,
383 json=True,
384 format="json",
385 )
386 defaults.update(kwargs)
387 return argparse.Namespace(**defaults)
388
389 def test_json_output_has_all_fields(self) -> None:
390 from muse.cli.commands.sign import run_payment
391
392 signing = _make_signing("alice")
393 args = self._args()
394 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
395 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
396 run_payment(args)
397 data = json.loads(mock_out.getvalue())
398 assert data["from_handle"] == "alice"
399 assert data["to_handle"] == "bob"
400 assert data["amount_nano"] == 1_000_000
401 assert data["currency"] == "nanoMUSE"
402 assert data["nonce_hex"] == self._NONCE
403 assert data["memo"] == "stem:sha256:abc123"
404 assert data["ts"] == 1744000000
405 assert "signature_b64" in data
406 assert "canonical_message" in data
407
408 def test_canonical_message_has_mpay_prefix(self) -> None:
409 """Payment canonical message must start with 'MPAY' domain separator."""
410 from muse.cli.commands.sign import run_payment
411
412 signing = _make_signing("alice")
413 args = self._args()
414 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
415 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
416 run_payment(args)
417 data = json.loads(mock_out.getvalue())
418 assert data["canonical_message"].startswith("MPAY\n")
419
420 def test_signature_is_verifiable(self) -> None:
421 """The payment signature must verify against the signer's public key."""
422 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
423 from muse.cli.commands.sign import run_payment
424 from muse.core.transport import SigningIdentity
425
426 private_key = Ed25519PrivateKey.generate()
427 signing = SigningIdentity(handle="alice", private_key=private_key)
428 args = self._args()
429 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
430 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
431 run_payment(args)
432 data = json.loads(mock_out.getvalue())
433
434 sig_bytes = base64.urlsafe_b64decode(data["signature_b64"] + "==")
435 msg = data["canonical_message"].encode()
436 # Must not raise InvalidSignature.
437 private_key.public_key().verify(sig_bytes, msg)
438
439 def test_domain_separation_from_http_msign(self) -> None:
440 """Payment signature must NOT verify against the HTTP MSign canonical message."""
441 from cryptography.exceptions import InvalidSignature
442 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
443 from muse.cli.commands.sign import run_payment
444 from muse.core.msign import canonical_message
445 from muse.core.transport import SigningIdentity
446
447 private_key = Ed25519PrivateKey.generate()
448 signing = SigningIdentity(handle="alice", private_key=private_key)
449 args = self._args()
450 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
451 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
452 run_payment(args)
453 data = json.loads(mock_out.getvalue())
454
455 sig_bytes = base64.urlsafe_b64decode(data["signature_b64"] + "==")
456 # Try to verify the payment sig against an HTTP canonical message — must fail.
457 http_msg = canonical_message("POST", "/alice/bob", 1744000000, b"", host="hub")
458 with self.assertRaises(InvalidSignature):
459 private_key.public_key().verify(sig_bytes, http_msg)
460
461 def test_deterministic_at_fixed_timestamp(self) -> None:
462 """Same key + same inputs + same ts → same signature every time."""
463 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
464 from muse.cli.commands.sign import run_payment
465 from muse.core.transport import SigningIdentity
466
467 private_key = Ed25519PrivateKey.generate()
468 signing = SigningIdentity(handle="alice", private_key=private_key)
469 args = self._args()
470
471 sigs: list[str] = []
472 for _ in range(3):
473 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
474 patch("sys.stdout", new_callable=io.StringIO) as mock_out:
475 run_payment(args)
476 sigs.append(json.loads(mock_out.getvalue())["signature_b64"])
477
478 assert sigs[0] == sigs[1] == sigs[2], "Payment signatures must be deterministic"
479
480 def test_text_output_prints_signature_to_stderr(self) -> None:
481 """Text mode must print payment info (including signature) to stderr only."""
482 from muse.cli.commands.sign import run_payment
483
484 signing = _make_signing("alice")
485 args = self._args(json=False, format="text")
486 with patch("muse.cli.commands.sign._load_signing", return_value=signing), \
487 patch("sys.stdout", new_callable=io.StringIO) as mock_out, \
488 patch("sys.stderr", new_callable=io.StringIO) as mock_err:
489 run_payment(args)
490 # stdout must be empty — signature no longer bleeds to stdout
491 assert mock_out.getvalue() == "", f"stdout must be empty in text mode, got: {mock_out.getvalue()!r}"
492 # stderr must contain the signature
493 stderr = mock_err.getvalue()
494 assert "Signature:" in stderr, f"'Signature:' missing from stderr: {stderr!r}"
495
496
497 # ---------------------------------------------------------------------------
498 # _load_signing — identity resolution
499 # ---------------------------------------------------------------------------
500
501 class TestLoadSigning(unittest.TestCase):
502 def test_exits_1_when_no_identity(self) -> None:
503 """When get_signing_identity returns None, must exit with code 1."""
504 from muse.cli.commands.sign import _load_signing
505
506 with patch("muse.cli.commands.sign.get_signing_identity", return_value=None, create=True):
507 # Patch the import inside _load_signing.
508 with patch("muse.cli.config.get_signing_identity", return_value=None):
509 with self.assertRaises(SystemExit) as cm:
510 _load_signing(hub="https://hub.example.com")
511 assert cm.exception.code == 1
512
513 def test_key_path_override_loads_from_file(self) -> None:
514 """--key-path must load the private key from the specified PEM file."""
515 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
516 from cryptography.hazmat.primitives.serialization import (
517 Encoding, NoEncryption, PrivateFormat,
518 )
519 from muse.cli.commands.sign import _load_signing
520 import tempfile
521
522 private_key = Ed25519PrivateKey.generate()
523 pem = private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())
524
525 with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
526 f.write(pem)
527 pem_path = f.name
528
529 try:
530 with patch("muse.core.keypair.load_private_key_from_pem", return_value=private_key):
531 signing = _load_signing(hub=None, key_path=pem_path)
532 assert signing is not None
533 # handle defaults to stem of filename.
534 import pathlib
535 assert signing.handle == pathlib.Path(pem_path).stem
536 finally:
537 import os
538 os.unlink(pem_path)
539
540
541 if __name__ == "__main__":
542 unittest.main()
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 139 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 142 days ago