gabriel / muse public
test_agent_key_fd.py python
490 lines 18.6 KB
Raw
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ breaking 10 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_I2_deterministic_key_from_same_seed(
102 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
103 ) -> None:
104 """I2: same sub-seed always produces the same signing key."""
105 from muse.cli.config import get_signing_identity
106
107 sub_seed = _make_sub_seed(account=2)
108
109 r_fd1 = _pipe_with_seed(sub_seed)
110 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd1))
111 result1 = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
112 try:
113 os.close(r_fd1)
114 except OSError:
115 pass
116
117 r_fd2 = _pipe_with_seed(sub_seed)
118 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd2))
119 result2 = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
120 try:
121 os.close(r_fd2)
122 except OSError:
123 pass
124
125 assert result1 is not None and result2 is not None
126 # Same key material → same public key bytes
127 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
128 pub1 = result1.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
129 pub2 = result2.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
130 assert pub1 == pub2
131
132 def test_I3_fd_closed_after_read(
133 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
134 ) -> None:
135 """I3: the fd is closed by get_signing_identity — cannot be read again."""
136 from muse.cli.config import get_signing_identity
137
138 sub_seed = _make_sub_seed(account=3)
139 r_fd = _pipe_with_seed(sub_seed)
140
141 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
142 get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
143
144 # fd must be closed
145 with pytest.raises(OSError):
146 os.read(r_fd, 1)
147
148 def test_I4_muse_agent_handle_sets_handle(
149 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
150 ) -> None:
151 """I4: MUSE_AGENT_HANDLE sets the identity handle."""
152 from muse.cli.config import get_signing_identity
153
154 sub_seed = _make_sub_seed(account=4)
155 r_fd = _pipe_with_seed(sub_seed)
156
157 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
158 monkeypatch.setenv("MUSE_AGENT_HANDLE", "my-agent-001")
159
160 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
161 try:
162 os.close(r_fd)
163 except OSError:
164 pass
165
166 assert result is not None
167 assert result.handle == "my-agent-001"
168
169 def test_I5_default_handle_is_agent(
170 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
171 ) -> None:
172 """I5: handle defaults to 'agent' when MUSE_AGENT_HANDLE is not set."""
173 from muse.cli.config import get_signing_identity
174
175 sub_seed = _make_sub_seed(account=5)
176 r_fd = _pipe_with_seed(sub_seed)
177
178 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
179 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
180
181 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
182 try:
183 os.close(r_fd)
184 except OSError:
185 pass
186
187 assert result is not None
188 assert result.handle == "agent"
189
190
191 # ---------------------------------------------------------------------------
192 # II Priority and fallback
193 # ---------------------------------------------------------------------------
194
195
196 class TestPriorityII:
197 def test_II1_fd_takes_priority_over_identity_store(
198 self,
199 monkeypatch: pytest.MonkeyPatch,
200 tmp_path: pathlib.Path,
201 ) -> None:
202 """II1: MUSE_AGENT_KEY_FD takes priority over file-based identity."""
203 import muse.core.identity as id_mod
204 # Patch resolve_signing_identity so we know if it was called
205 called = []
206 orig = id_mod.resolve_signing_identity
207 monkeypatch.setattr(id_mod, "resolve_signing_identity",
208 lambda *a, **kw: (called.append(True), orig(*a, **kw))[1])
209
210 from muse.cli.config import get_signing_identity
211 sub_seed = _make_sub_seed(account=6)
212 r_fd = _pipe_with_seed(sub_seed)
213 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
214
215 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
216 try:
217 os.close(r_fd)
218 except OSError:
219 pass
220
221 assert result is not None
222 assert not called, "identity store was consulted despite MUSE_AGENT_KEY_FD being set"
223
224 def test_II2_falls_through_to_identity_store_when_unset(
225 self,
226 monkeypatch: pytest.MonkeyPatch,
227 tmp_path: pathlib.Path,
228 ) -> None:
229 """II2: without MUSE_AGENT_KEY_FD, falls through to identity store (returns None)."""
230 from muse.cli.config import get_signing_identity
231 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
232 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
233
234 # No identity configured → returns None
235 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
236 assert result is None
237
238
239 # ---------------------------------------------------------------------------
240 # III Error handling
241 # ---------------------------------------------------------------------------
242
243
244 class TestErrorHandlingIII:
245 def test_III1_invalid_fd_falls_through(
246 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
247 ) -> None:
248 """III1: an invalid fd number falls through gracefully (no crash)."""
249 from muse.cli.config import get_signing_identity
250 monkeypatch.setenv("MUSE_AGENT_KEY_FD", "9999") # certainly not open
251
252 # Should not raise — just falls through to identity store (returns None)
253 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
254 assert result is None
255
256 def test_III2_wrong_byte_count_falls_through(
257 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
258 ) -> None:
259 """III2: fewer than 64 bytes in the pipe → falls through."""
260 from muse.cli.config import get_signing_identity
261
262 r_fd, w_fd = os.pipe()
263 os.write(w_fd, b"\x00" * 32) # 32 bytes, not 64
264 os.close(w_fd)
265
266 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
267
268 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
269 try:
270 os.close(r_fd)
271 except OSError:
272 pass
273
274 assert result is None
275
276 def test_III3_muse_agent_hd_seed_not_recognised(
277 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
278 ) -> None:
279 """III3: MUSE_AGENT_HD_SEED is no longer supported — setting it has no effect."""
280 from muse.cli.config import get_signing_identity
281 from muse.core.bip39 import mnemonic_to_seed
282 from muse.core.hdkeys import DOMAIN_IDENTITY, derive_agent_sub_seed
283 from muse.core.types import b64url_encode
284
285 sub_seed = _make_sub_seed(account=7)
286 seed_b64 = b64url_encode(sub_seed)
287
288 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
289 monkeypatch.setenv("MUSE_AGENT_HD_SEED", seed_b64)
290
291 # Should NOT return a signing identity (env var is removed)
292 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
293 assert result is None, (
294 "MUSE_AGENT_HD_SEED should be ignored but returned a signing identity"
295 )
296
297 def test_III4_muse_agent_key_not_recognised(
298 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
299 ) -> None:
300 """III4: MUSE_AGENT_KEY (PEM env var) is no longer supported."""
301 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
302 from cryptography.hazmat.primitives.serialization import (
303 Encoding, PrivateFormat, NoEncryption,
304 )
305 from muse.cli.config import get_signing_identity
306
307 key = Ed25519PrivateKey.generate()
308 pem = key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()).decode()
309
310 monkeypatch.delenv("MUSE_AGENT_KEY_FD", raising=False)
311 monkeypatch.setenv("MUSE_AGENT_KEY", pem)
312
313 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
314 assert result is None, (
315 "MUSE_AGENT_KEY should be ignored but returned a signing identity"
316 )
317
318
319 # ---------------------------------------------------------------------------
320 # IV Security
321 # ---------------------------------------------------------------------------
322
323
324 class TestHubScopingV:
325 """V — musehub#221: fd-injected keys must be hub-scoped, never derived blind."""
326
327 def test_V1_no_resolvable_hub_falls_through_without_deriving(
328 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
329 ) -> None:
330 """No remote_url and no [hub] url in config.toml → the fd branch must
331 not derive a hub-independent key; it falls through and, with no
332 identity-store hub either, returns None."""
333 from muse.cli.config import get_signing_identity
334 from muse.core import hdkeys as _hdkeys
335 from unittest.mock import patch
336
337 sub_seed = _make_sub_seed(account=42)
338 r_fd = _pipe_with_seed(sub_seed)
339 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
340
341 called = []
342 with patch.object(_hdkeys, "derive_identity_key", side_effect=called.append):
343 # No remote_url, and tmp_path has no .muse/config.toml [hub] url.
344 result = get_signing_identity(repo_root=tmp_path)
345
346 try:
347 os.close(r_fd)
348 except OSError:
349 pass
350
351 assert not called, "derive_identity_key must never be called without a resolvable hub"
352 assert result is None
353
354 def test_V2_resolvable_hub_derives_scoped_to_it(
355 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
356 ) -> None:
357 """Same sub-seed, two different remote_url values, must yield two
358 different signing keys — the exact property musehub#221 requires."""
359 from muse.cli.config import get_signing_identity
360 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
361
362 sub_seed = _make_sub_seed(account=43)
363
364 r_fd_a = _pipe_with_seed(sub_seed)
365 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_a))
366 result_a = get_signing_identity(repo_root=tmp_path, remote_url="https://musehub.ai")
367 try:
368 os.close(r_fd_a)
369 except OSError:
370 pass
371
372 r_fd_b = _pipe_with_seed(sub_seed)
373 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_b))
374 result_b = get_signing_identity(repo_root=tmp_path, remote_url="https://staging.musehub.ai")
375 try:
376 os.close(r_fd_b)
377 except OSError:
378 pass
379
380 assert result_a is not None and result_b is not None
381 pub_a = result_a.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
382 pub_b = result_b.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
383 assert pub_a != pub_b
384
385
386 class TestSecurityIV:
387 def test_IV1_sub_seed_not_in_environ(
388 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
389 ) -> None:
390 """IV1: the sub-seed bytes never appear in os.environ."""
391 from muse.cli.config import get_signing_identity
392
393 sub_seed = _make_sub_seed(account=8)
394 r_fd = _pipe_with_seed(sub_seed)
395 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
396
397 get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
398 try:
399 os.close(r_fd)
400 except OSError:
401 pass
402
403 # The raw bytes and any base64 encoding of them must not be in environ
404 from muse.core.types import b64url_encode
405 seed_b64 = b64url_encode(sub_seed)
406 for val in os.environ.values():
407 assert seed_b64 not in val, "sub-seed base64 found in os.environ"
408
409 def test_IV2_different_seeds_different_keys(
410 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
411 ) -> None:
412 """IV2: two different sub-seeds produce two different signing keys."""
413 from muse.cli.config import get_signing_identity
414 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
415
416 sub_seed_a = _make_sub_seed(account=9)
417 sub_seed_b = _make_sub_seed(account=10)
418
419 r_fd_a = _pipe_with_seed(sub_seed_a)
420 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_a))
421 result_a = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
422 try:
423 os.close(r_fd_a)
424 except OSError:
425 pass
426
427 r_fd_b = _pipe_with_seed(sub_seed_b)
428 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd_b))
429 result_b = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
430 try:
431 os.close(r_fd_b)
432 except OSError:
433 pass
434
435 assert result_a is not None and result_b is not None
436 pub_a = result_a.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
437 pub_b = result_b.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
438 assert pub_a != pub_b, "Different sub-seeds produced identical keys"
439
440 def test_IV3_sub_seed_zeroed_after_use(
441 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
442 ) -> None:
443 """IV3: the sub-seed buffer must be zeroed in memory after key derivation.
444
445 CRITICAL-2: a lingering plaintext sub-seed in RAM can be recovered from
446 a core dump or via /proc/<pid>/mem. The buffer must be all-zero before
447 get_signing_identity returns.
448
449 Strategy: patch muse.core.hdkeys.derive_identity_key to capture a
450 reference to the buffer passed by the caller. After get_signing_identity
451 returns we verify:
452 1. The buffer is a bytearray (mutable, so it *can* be zeroed).
453 2. Every byte is 0x00 (was zeroed before returning).
454 """
455 import os
456 from unittest.mock import patch
457 from muse.cli.config import get_signing_identity
458 from muse.core import hdkeys as _hdkeys
459
460 sub_seed = _make_sub_seed(account=99)
461 r_fd = _pipe_with_seed(sub_seed)
462 monkeypatch.setenv("MUSE_AGENT_KEY_FD", str(r_fd))
463 monkeypatch.delenv("MUSE_AGENT_HANDLE", raising=False)
464
465 captured = []
466 original_derive = _hdkeys.derive_identity_key
467
468 def capturing_derive(seed: bytes, *, hub: int) -> DerivedKey:
469 captured.append(seed) # keep a reference — will survive zeroing
470 return original_derive(seed, hub=hub)
471
472 with patch.object(_hdkeys, "derive_identity_key", side_effect=capturing_derive):
473 result = get_signing_identity(repo_root=tmp_path, remote_url=_TEST_HUB)
474
475 try:
476 os.close(r_fd)
477 except OSError:
478 pass
479
480 assert result is not None, "get_signing_identity must still succeed"
481 assert len(captured) == 1, "derive_identity_key must be called exactly once"
482
483 buf = captured[0]
484 assert isinstance(buf, bytearray), (
485 f"sub-seed must be passed as bytearray (got {type(buf).__name__}) "
486 "so it can be zeroed after use"
487 )
488 assert buf == bytearray(64), (
489 "sub-seed buffer must be all-zero after get_signing_identity returns"
490 )
File History 1 commit
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ 10 hours ago