gabriel / muse public
test_agent_key_fd.py python
526 lines 20.1 KB
Raw
sha256:e6186c2a5be8e7a1361b66ad64d74d3ed1865f3e713c55e7931b4c983e3c9df6 fix: get_signing_identity's MUSE_AGENT_KEY_FD path derived … Sonnet 5 patch 5 hours ago
1 """Tests for fd-based agent key injection — Tier 3.
2
3 MUSE_AGENT_KEY_FD is the only supported env-var mechanism for injecting
4 a sub-seed into an agent subprocess. The old MUSE_AGENT_HD_SEED and
5 MUSE_AGENT_KEY env vars are removed.
6
7 Protocol:
8 1. Parent creates an anonymous pipe (r_fd, w_fd).
9 2. Parent writes exactly 64 bytes of sub-seed to w_fd, closes w_fd.
10 3. Parent sets MUSE_AGENT_KEY_FD=str(r_fd) and spawns child with pass_fds=(r_fd,).
11 4. Child (get_signing_identity) reads exactly 64 bytes from r_fd, closes r_fd.
12 5. Child derives Ed25519 private key from sub_seed via derive_identity_key.
13 6. Secret never appears in /proc/<pid>/environ.
14
15 Coverage
16 --------
17 I get_signing_identity — fd injection
18 I1 MUSE_AGENT_KEY_FD reads 64 bytes, returns valid SigningIdentity
19 I2 derived key is deterministic for the same sub-seed
20 I3 fd is closed after read (cannot be read a second time)
21 I4 MUSE_AGENT_HANDLE sets the identity handle
22 I5 handle defaults to "agent" when MUSE_AGENT_HANDLE is unset
23
24 II Priority and fallback
25 II1 MUSE_AGENT_KEY_FD takes priority over identity store
26 II2 falls through to identity store when MUSE_AGENT_KEY_FD is unset
27
28 III Error handling
29 III1 invalid fd number → falls through (does not crash)
30 III2 fd with wrong byte count → falls through
31 III3 MUSE_AGENT_HD_SEED is no longer recognised
32 III4 MUSE_AGENT_KEY is no longer recognised
33
34 IV Security
35 IV1 sub-seed does not appear in any log output
36 IV2 two different sub-seeds produce two different signing keys
37 """
38
39 from __future__ import annotations
40
41 import os
42 import pathlib
43
44 import pytest
45 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
46 from muse.core.slip010 import DerivedKey
47
48 _TEST_MNEMONIC = (
49 "abandon abandon abandon abandon abandon abandon abandon abandon "
50 "abandon abandon abandon about"
51 )
52
53 # musehub#221 — get_signing_identity requires a resolvable hub to derive an
54 # fd-injected key; every real caller (push/pull/hub commands) always has one.
55 _TEST_HUB = "https://musehub.ai"
56
57
58 def _make_sub_seed(account: int = 1) -> bytes:
59 """Derive a real 64-byte IDENTITY-domain agent sub-seed."""
60 from muse.core.bip39 import mnemonic_to_seed
61 from muse.core.hdkeys import DOMAIN_IDENTITY, derive_agent_sub_seed
62 seed = mnemonic_to_seed(_TEST_MNEMONIC)
63 return derive_agent_sub_seed(seed, domain=DOMAIN_IDENTITY, agent_id=account)
64
65
66 def _pipe_with_seed(sub_seed: bytes) -> int:
67 """Create a pipe, write sub_seed, close write end, return read fd."""
68 r_fd, w_fd = os.pipe()
69 os.write(w_fd, sub_seed)
70 os.close(w_fd)
71 return r_fd
72
73
74 # ---------------------------------------------------------------------------
75 # I get_signing_identity — fd injection
76 # ---------------------------------------------------------------------------
77
78
79 class TestFdInjectionI:
80 def test_I1_fd_returns_signing_identity(
81 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
82 ) -> None:
83 """I1: MUSE_AGENT_KEY_FD yields a valid SigningIdentity."""
84 from muse.cli.config import get_signing_identity
85
86 sub_seed = _make_sub_seed(account=1)
87 r_fd = _pipe_with_seed(sub_seed)
88
89 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
90 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
91
92 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
93 try:
94 os.close(r_fd)
95 except OSError:
96 pass
97
98 assert result is not None
99 assert isinstance(result.private_key, Ed25519PrivateKey)
100
101 def test_I6_derives_with_entity_type_agent(
102 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
103 ) -> None:
104 """musehub#205: the MUSE_AGENT_KEY_FD path must derive the identity
105 key with entity_type=ENTITY_AGENT, matching
106 derive_identity_key's own documented example for an agent sub-seed
107 ("Agent slot 0's identity key"). The previous default
108 (entity_type=ENTITY_HUMAN) rooted every agent key under the wrong
109 entity branch of the agent's own already-isolated sub-tree."""
110 from unittest.mock import patch
111 from muse.cli.config import get_signing_identity
112 from muse.core import hdkeys as _hdkeys
113
114 sub_seed = _make_sub_seed(account=7)
115 r_fd = _pipe_with_seed(sub_seed)
116 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
117 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
118
119 captured_kwargs = {}
120 original_derive = _hdkeys.derive_identity_key
121
122 def capturing_derive(seed: bytes, **kwargs: object) -> DerivedKey:
123 captured_kwargs.update(kwargs)
124 return original_derive(seed, **kwargs)
125
126 with patch.object(_hdkeys, "derive_identity_key", side_effect=capturing_derive):
127 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
128
129 try:
130 os.close(r_fd)
131 except OSError:
132 pass
133
134 assert result is not None
135 assert captured_kwargs.get("entity_type") == _hdkeys.ENTITY_AGENT
136
137 def test_I2_deterministic_key_from_same_seed(
138 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
139 ) -> None:
140 """I2: same sub-seed always produces the same signing key."""
141 from muse.cli.config import get_signing_identity
142
143 sub_seed = _make_sub_seed(account=2)
144
145 r_fd1 = _pipe_with_seed(sub_seed)
146 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd1))
147 result1 = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
148 try:
149 os.close(r_fd1)
150 except OSError:
151 pass
152
153 r_fd2 = _pipe_with_seed(sub_seed)
154 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd2))
155 result2 = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
156 try:
157 os.close(r_fd2)
158 except OSError:
159 pass
160
161 assert result1 is not None and result2 is not None
162 # Same key material → same public key bytes
163 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
164 pub1 = result1.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
165 pub2 = result2.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
166 assert pub1 == pub2
167
168 def test_I3_fd_closed_after_read(
169 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
170 ) -> None:
171 """I3: the fd is closed by get_signing_identity — cannot be read again."""
172 from muse.cli.config import get_signing_identity
173
174 sub_seed = _make_sub_seed(account=3)
175 r_fd = _pipe_with_seed(sub_seed)
176
177 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
178 get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
179
180 # fd must be closed
181 with pytest.raises(OSError):
182 os.read(r_fd, 1)
183
184 def test_I4_muse_agent_handle_sets_handle(
185 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
186 ) -> None:
187 """I4: MUSE_AGENT_HANDLE sets the identity handle."""
188 from muse.cli.config import get_signing_identity
189
190 sub_seed = _make_sub_seed(account=4)
191 r_fd = _pipe_with_seed(sub_seed)
192
193 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
194 monkeypatch.setenv("MUSE_AGENT_HANDLE", "my-agent-001")
195
196 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
197 try:
198 os.close(r_fd)
199 except OSError:
200 pass
201
202 assert result is not None
203 assert result.handle == "my-agent-001"
204
205 def test_I5_default_handle_is_agent(
206 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
207 ) -> None:
208 """I5: handle defaults to 'agent' when MUSE_AGENT_HANDLE is not set."""
209 from muse.cli.config import get_signing_identity
210
211 sub_seed = _make_sub_seed(account=5)
212 r_fd = _pipe_with_seed(sub_seed)
213
214 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
215 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
216
217 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
218 try:
219 os.close(r_fd)
220 except OSError:
221 pass
222
223 assert result is not None
224 assert result.handle == "agent"
225
226
227 # ---------------------------------------------------------------------------
228 # II Priority and fallback
229 # ---------------------------------------------------------------------------
230
231
232 class TestPriorityII:
233 def test_II1_fd_takes_priority_over_identity_store(
234 self,
235 monkeypatch: pytest.MonkeyPatch,
236 tmp_path: pathlib.Path,
237 ) -> None:
238 """II1: MUSE_AGENT_KEY_FD takes priority over file-based identity."""
239 import muse.core.identity as id_mod
240 # Patch resolve_signing_identity so we know if it was called
241 called = []
242 orig = id_mod.resolve_signing_identity
243 monkeypatch.setattr(id_mod, "resolve_signing_identity",
244 lambda *a, **kw: (called.append(True), orig(*a, **kw))[1])
245
246 from muse.cli.config import get_signing_identity
247 sub_seed = _make_sub_seed(account=6)
248 r_fd = _pipe_with_seed(sub_seed)
249 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
250
251 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
252 try:
253 os.close(r_fd)
254 except OSError:
255 pass
256
257 assert result is not None
258 assert not called, "identity store was consulted despite MUSE_AGENT_KEY_FD being set"
259
260 def test_II2_falls_through_to_identity_store_when_unset(
261 self,
262 monkeypatch: pytest.MonkeyPatch,
263 tmp_path: pathlib.Path,
264 ) -> None:
265 """II2: without MUSE_AGENT_KEY_FD, falls through to identity store (returns None)."""
266 from muse.cli.config import get_signing_identity
267 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
268 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
269
270 # No identity configured → returns None
271 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
272 assert result is None
273
274
275 # ---------------------------------------------------------------------------
276 # III Error handling
277 # ---------------------------------------------------------------------------
278
279
280 class TestErrorHandlingIII:
281 def test_III1_invalid_fd_falls_through(
282 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
283 ) -> None:
284 """III1: an invalid fd number falls through gracefully (no crash)."""
285 from muse.cli.config import get_signing_identity
286 monkeypatch.setenv("MUSE_AGENT_KEY_FD", "9999") # certainly not open
287
288 # Should not raise — just falls through to identity store (returns None)
289 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
290 assert result is None
291
292 def test_III2_wrong_byte_count_falls_through(
293 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
294 ) -> None:
295 """III2: fewer than 64 bytes in the pipe → falls through."""
296 from muse.cli.config import get_signing_identity
297
298 r_fd, w_fd = os.pipe()
299 os.write(w_fd, b"\x00" * 32) # 32 bytes, not 64
300 os.close(w_fd)
301
302 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
303
304 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
305 try:
306 os.close(r_fd)
307 except OSError:
308 pass
309
310 assert result is None
311
312 def test_III3_muse_agent_hd_seed_not_recognised(
313 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
314 ) -> None:
315 """III3: MUSE_AGENT_HD_SEED is no longer supported — setting it has no effect."""
316 from muse.cli.config import get_signing_identity
317 from muse.core.bip39 import mnemonic_to_seed
318 from muse.core.hdkeys import DOMAIN_IDENTITY, derive_agent_sub_seed
319 from muse.core.types import b64url_encode
320
321 sub_seed = _make_sub_seed(account=7)
322 seed_b64 = b64url_encode(sub_seed)
323
324 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
325 monkeypatch.setenv("MUSE_AGENT_HD_SEED", seed_b64)
326
327 # Should NOT return a signing identity (env var is removed)
328 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
329 assert result is None, (
330 "MUSE_AGENT_HD_SEED should be ignored but returned a signing identity"
331 )
332
333 def test_III4_muse_agent_key_not_recognised(
334 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
335 ) -> None:
336 """III4: MUSE_AGENT_KEY (PEM env var) is no longer supported."""
337 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
338 from cryptography.hazmat.primitives.serialization import (
339 Encoding, PrivateFormat, NoEncryption,
340 )
341 from muse.cli.config import get_signing_identity
342
343 key = Ed25519PrivateKey.generate()
344 pem = key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()).decode()
345
346 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
347 monkeypatch.setenv("MUSE_AGENT_KEY", pem)
348
349 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
350 assert result is None, (
351 "MUSE_AGENT_KEY should be ignored but returned a signing identity"
352 )
353
354
355 # ---------------------------------------------------------------------------
356 # IV Security
357 # ---------------------------------------------------------------------------
358
359
360 class TestHubScopingV:
361 """V — musehub#221: fd-injected keys must be hub-scoped, never derived blind."""
362
363 def test_V1_no_resolvable_hub_falls_through_without_deriving(
364 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
365 ) -> None:
366 """No remote_url and no [hub] url in config.toml → the fd branch must
367 not derive a hub-independent key; it falls through and, with no
368 identity-store hub either, returns None."""
369 from muse.cli.config import get_signing_identity
370 from muse.core import hdkeys as _hdkeys
371 from unittest.mock import patch
372
373 sub_seed = _make_sub_seed(account=42)
374 r_fd = _pipe_with_seed(sub_seed)
375 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
376
377 called = []
378 with patch.object(_hdkeys, "derive_identity_key", side_effect=called.append):
379 # No remote_url, and tmp_path has no .muse/config.toml [hub] url.
380 result = get_signing_identity(repo_root=tmp_path)
381
382 try:
383 os.close(r_fd)
384 except OSError:
385 pass
386
387 assert not called, "derive_identity_key must never be called without a resolvable hub"
388 assert result is None
389
390 def test_V2_resolvable_hub_derives_scoped_to_it(
391 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
392 ) -> None:
393 """Same sub-seed, two different remote_url values, must yield two
394 different signing keys — the exact property musehub#221 requires."""
395 from muse.cli.config import get_signing_identity
396 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
397
398 sub_seed = _make_sub_seed(account=43)
399
400 r_fd_a = _pipe_with_seed(sub_seed)
401 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_a))
402 result_a = get_signing_identity(repo_root=tmp_path, remote_url="https://musehub.ai")
403 try:
404 os.close(r_fd_a)
405 except OSError:
406 pass
407
408 r_fd_b = _pipe_with_seed(sub_seed)
409 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_b))
410 result_b = get_signing_identity(repo_root=tmp_path, remote_url="https://staging.musehub.ai")
411 try:
412 os.close(r_fd_b)
413 except OSError:
414 pass
415
416 assert result_a is not None and result_b is not None
417 pub_a = result_a.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
418 pub_b = result_b.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
419 assert pub_a != pub_b
420
421
422 class TestSecurityIV:
423 def test_IV1_sub_seed_not_in_environ(
424 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
425 ) -> None:
426 """IV1: the sub-seed bytes never appear in os.environ."""
427 from muse.cli.config import get_signing_identity
428
429 sub_seed = _make_sub_seed(account=8)
430 r_fd = _pipe_with_seed(sub_seed)
431 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
432
433 get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
434 try:
435 os.close(r_fd)
436 except OSError:
437 pass
438
439 # The raw bytes and any base64 encoding of them must not be in environ
440 from muse.core.types import b64url_encode
441 seed_b64 = b64url_encode(sub_seed)
442 for val in os.environ.values():
443 assert seed_b64 not in val, "sub-seed base64 found in os.environ"
444
445 def test_IV2_different_seeds_different_keys(
446 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
447 ) -> None:
448 """IV2: two different sub-seeds produce two different signing keys."""
449 from muse.cli.config import get_signing_identity
450 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
451
452 sub_seed_a = _make_sub_seed(account=9)
453 sub_seed_b = _make_sub_seed(account=10)
454
455 r_fd_a = _pipe_with_seed(sub_seed_a)
456 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_a))
457 result_a = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
458 try:
459 os.close(r_fd_a)
460 except OSError:
461 pass
462
463 r_fd_b = _pipe_with_seed(sub_seed_b)
464 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_b))
465 result_b = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
466 try:
467 os.close(r_fd_b)
468 except OSError:
469 pass
470
471 assert result_a is not None and result_b is not None
472 pub_a = result_a.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
473 pub_b = result_b.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
474 assert pub_a != pub_b, "Different sub-seeds produced identical keys"
475
476 def test_IV3_sub_seed_zeroed_after_use(
477 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
478 ) -> None:
479 """IV3: the sub-seed buffer must be zeroed in memory after key derivation.
480
481 CRITICAL-2: a lingering plaintext sub-seed in RAM can be recovered from
482 a core dump or via /proc/<pid>/mem. The buffer must be all-zero before
483 get_signing_identity returns.
484
485 Strategy: patch muse.core.hdkeys.derive_identity_key to capture a
486 reference to the buffer passed by the caller. After get_signing_identity
487 returns we verify:
488 1. The buffer is a bytearray (mutable, so it *can* be zeroed).
489 2. Every byte is 0x00 (was zeroed before returning).
490 """
491 import os
492 from unittest.mock import patch
493 from muse.cli.config import get_signing_identity
494 from muse.core import hdkeys as _hdkeys
495
496 sub_seed = _make_sub_seed(account=99)
497 r_fd = _pipe_with_seed(sub_seed)
498 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
499 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
500
501 captured = []
502 original_derive = _hdkeys.derive_identity_key
503
504 def capturing_derive(seed: bytes, **kwargs: object) -> DerivedKey:
505 captured.append(seed) # keep a reference — will survive zeroing
506 return original_derive(seed, **kwargs)
507
508 with patch.object(_hdkeys, "derive_identity_key", side_effect=capturing_derive):
509 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
510
511 try:
512 os.close(r_fd)
513 except OSError:
514 pass
515
516 assert result is not None, "get_signing_identity must still succeed"
517 assert len(captured) == 1, "derive_identity_key must be called exactly once"
518
519 buf = captured[0]
520 assert isinstance(buf, bytearray), (
521 f"sub-seed must be passed as bytearray (got {type(buf).__name__}) "
522 "so it can be zeroed after use"
523 )
524 assert buf == bytearray(64), (
525 "sub-seed buffer must be all-zero after get_signing_identity returns"
526 )
File History 1 commit
sha256:e6186c2a5be8e7a1361b66ad64d74d3ed1865f3e713c55e7931b4c983e3c9df6 fix: get_signing_identity's MUSE_AGENT_KEY_FD path derived … Sonnet 5 patch 5 hours ago