test_agent_signing.py
python
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3
docs: docstring sprint contract→find-symbol — idiomatic run…
Sonnet 4.6
patch
139 days ago
| 1 | """Tests for agent-first signing — compound identity keys and key paths. |
| 2 | |
| 3 | Covers: |
| 4 | - identity.py: compound key load/save/clear, provisioned_by field |
| 5 | - keypair.py: agent-specific key paths and HD key generation |
| 6 | - resolve_signing_identity: agent key → human key fallback |
| 7 | """ |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import pathlib |
| 11 | import json |
| 12 | |
| 13 | import pytest |
| 14 | |
| 15 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 16 | from cryptography.hazmat.primitives.serialization import ( |
| 17 | Encoding, |
| 18 | NoEncryption, |
| 19 | PrivateFormat, |
| 20 | ) |
| 21 | |
| 22 | from muse.core.identity import ( |
| 23 | IdentityEntry, |
| 24 | _identity_key, |
| 25 | clear_identity, |
| 26 | hostname_from_url, |
| 27 | list_all_identities, |
| 28 | load_identity, |
| 29 | resolve_signing_identity, |
| 30 | save_identity, |
| 31 | ) |
| 32 | from muse.core.keypair import ( |
| 33 | generate_hd_keypair, |
| 34 | key_path_for, |
| 35 | load_private_key, |
| 36 | load_private_key_from_pem, |
| 37 | ) |
| 38 | |
| 39 | # Fixed 64-byte seeds for deterministic test keys |
| 40 | _SEED_A = b"\x00" * 64 |
| 41 | _SEED_B = b"\x01" * 64 |
| 42 | |
| 43 | |
| 44 | # --------------------------------------------------------------------------- |
| 45 | # Fixtures |
| 46 | # --------------------------------------------------------------------------- |
| 47 | |
| 48 | |
| 49 | @pytest.fixture() |
| 50 | def isolated_identity(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 51 | """Redirect the identity store to a temp directory for test isolation.""" |
| 52 | import muse.core.identity as _id_mod |
| 53 | identity_dir = tmp_path / ".muse" |
| 54 | identity_dir.mkdir() |
| 55 | monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", identity_dir) |
| 56 | monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", identity_dir / "identity.toml") |
| 57 | return identity_dir |
| 58 | |
| 59 | |
| 60 | @pytest.fixture() |
| 61 | def isolated_keys(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 62 | """Redirect key storage to a temp directory.""" |
| 63 | import muse.core.keypair as _kp_mod |
| 64 | keys_dir = tmp_path / ".muse" / "keys" |
| 65 | keys_dir.mkdir(parents=True) |
| 66 | monkeypatch.setattr(_kp_mod, "_KEYS_DIR", keys_dir) |
| 67 | return keys_dir |
| 68 | |
| 69 | |
| 70 | # --------------------------------------------------------------------------- |
| 71 | # _identity_key helper |
| 72 | # --------------------------------------------------------------------------- |
| 73 | |
| 74 | |
| 75 | class TestIdentityKey: |
| 76 | def test_human_key_is_bare_hostname(self) -> None: |
| 77 | assert _identity_key("localhost:10003") == "localhost:10003" |
| 78 | |
| 79 | def test_agent_key_uses_hash_separator(self) -> None: |
| 80 | assert _identity_key("localhost:10003", "agent-abc") == "localhost:10003#agent-abc" |
| 81 | |
| 82 | def test_none_agent_id_gives_bare_hostname(self) -> None: |
| 83 | assert _identity_key("musehub.ai", None) == "musehub.ai" |
| 84 | |
| 85 | def test_empty_agent_id_gives_bare_hostname(self) -> None: |
| 86 | # empty string is falsy |
| 87 | assert _identity_key("musehub.ai", "") == "musehub.ai" |
| 88 | |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # load_identity / save_identity compound keys |
| 92 | # --------------------------------------------------------------------------- |
| 93 | |
| 94 | |
| 95 | class TestCompoundIdentityKeys: |
| 96 | def test_human_and_agent_entries_coexist( |
| 97 | self, isolated_identity: pathlib.Path |
| 98 | ) -> None: |
| 99 | hub = "http://localhost:10003" |
| 100 | |
| 101 | human: IdentityEntry = { |
| 102 | "type": "human", |
| 103 | "handle": "gabriel", |
| 104 | "key_path": "/fake/human.pem", |
| 105 | "algorithm": "ed25519", |
| 106 | "fingerprint": "a" * 64, |
| 107 | } |
| 108 | agent: IdentityEntry = { |
| 109 | "type": "agent", |
| 110 | "handle": "agentception-abc", |
| 111 | "key_path": "/fake/agent.pem", |
| 112 | "algorithm": "ed25519", |
| 113 | "fingerprint": "b" * 64, |
| 114 | "provisioned_by": "gabriel", |
| 115 | } |
| 116 | |
| 117 | save_identity(hub, human) |
| 118 | save_identity(hub, agent, agent_id="agentception-abc") |
| 119 | |
| 120 | loaded_human = load_identity(hub) |
| 121 | loaded_agent = load_identity(hub, agent_id="agentception-abc") |
| 122 | |
| 123 | assert loaded_human is not None |
| 124 | assert loaded_human["handle"] == "gabriel" |
| 125 | assert loaded_human["type"] == "human" |
| 126 | |
| 127 | assert loaded_agent is not None |
| 128 | assert loaded_agent["handle"] == "agentception-abc" |
| 129 | assert loaded_agent["type"] == "agent" |
| 130 | assert loaded_agent.get("provisioned_by") == "gabriel" |
| 131 | |
| 132 | def test_agent_entry_does_not_shadow_human( |
| 133 | self, isolated_identity: pathlib.Path |
| 134 | ) -> None: |
| 135 | hub = "http://localhost:10003" |
| 136 | human: IdentityEntry = {"type": "human", "handle": "gabriel"} |
| 137 | save_identity(hub, human) |
| 138 | |
| 139 | # Load without agent_id → human entry |
| 140 | loaded = load_identity(hub) |
| 141 | assert loaded is not None |
| 142 | assert loaded["handle"] == "gabriel" |
| 143 | |
| 144 | def test_load_missing_agent_returns_none( |
| 145 | self, isolated_identity: pathlib.Path |
| 146 | ) -> None: |
| 147 | hub = "http://localhost:10003" |
| 148 | assert load_identity(hub, agent_id="nonexistent") is None |
| 149 | |
| 150 | def test_clear_agent_identity_leaves_human_intact( |
| 151 | self, isolated_identity: pathlib.Path |
| 152 | ) -> None: |
| 153 | hub = "http://localhost:10003" |
| 154 | human: IdentityEntry = {"type": "human", "handle": "gabriel"} |
| 155 | agent: IdentityEntry = {"type": "agent", "handle": "agentception-abc", "provisioned_by": "gabriel"} |
| 156 | |
| 157 | save_identity(hub, human) |
| 158 | save_identity(hub, agent, agent_id="agentception-abc") |
| 159 | |
| 160 | cleared = clear_identity(hub, agent_id="agentception-abc") |
| 161 | assert cleared is True |
| 162 | |
| 163 | assert load_identity(hub) is not None # human still present |
| 164 | assert load_identity(hub, agent_id="agentception-abc") is None # agent gone |
| 165 | |
| 166 | def test_provisioned_by_roundtrip( |
| 167 | self, isolated_identity: pathlib.Path |
| 168 | ) -> None: |
| 169 | """provisioned_by survives a save/load cycle.""" |
| 170 | hub = "http://localhost:10003" |
| 171 | agent: IdentityEntry = { |
| 172 | "type": "agent", |
| 173 | "handle": "bot-001", |
| 174 | "provisioned_by": "gabriel", |
| 175 | "algorithm": "ed25519", |
| 176 | "fingerprint": "c" * 64, |
| 177 | } |
| 178 | save_identity(hub, agent, agent_id="bot-001") |
| 179 | loaded = load_identity(hub, agent_id="bot-001") |
| 180 | assert loaded is not None |
| 181 | assert loaded.get("provisioned_by") == "gabriel" |
| 182 | |
| 183 | def test_list_all_includes_compound_keys( |
| 184 | self, isolated_identity: pathlib.Path |
| 185 | ) -> None: |
| 186 | hub = "http://localhost:10003" |
| 187 | save_identity(hub, {"type": "human", "handle": "gabriel"}) |
| 188 | save_identity(hub, {"type": "agent", "handle": "bot"}, agent_id="bot") |
| 189 | |
| 190 | all_ids = list_all_identities() |
| 191 | assert "localhost:10003" in all_ids |
| 192 | assert "localhost:10003#bot" in all_ids |
| 193 | |
| 194 | |
| 195 | # --------------------------------------------------------------------------- |
| 196 | # keypair agent-specific paths |
| 197 | # --------------------------------------------------------------------------- |
| 198 | |
| 199 | |
| 200 | class TestAgentKeyPaths: |
| 201 | def test_human_key_path(self) -> None: |
| 202 | p = key_path_for("localhost:10003") |
| 203 | assert "__" not in p.name |
| 204 | assert p.name == "localhost_10003.pem" |
| 205 | |
| 206 | def test_agent_key_path_uses_double_underscore(self) -> None: |
| 207 | p = key_path_for("localhost:10003", "agentception-abc") |
| 208 | assert p.name == "localhost_10003__agentception-abc.pem" |
| 209 | |
| 210 | def test_generate_agent_keypair_creates_distinct_file( |
| 211 | self, isolated_keys: pathlib.Path |
| 212 | ) -> None: |
| 213 | pub_human, fp_human = generate_hd_keypair("localhost:10003", _SEED_A) |
| 214 | pub_agent, fp_agent = generate_hd_keypair("localhost:10003", _SEED_B, "agentception-abc") |
| 215 | |
| 216 | assert fp_human != fp_agent # different seeds → different keys |
| 217 | assert (isolated_keys / "localhost_10003.pem").is_file() |
| 218 | assert (isolated_keys / "localhost_10003__agentception-abc.pem").is_file() |
| 219 | |
| 220 | def test_load_private_key_with_agent_id( |
| 221 | self, isolated_keys: pathlib.Path |
| 222 | ) -> None: |
| 223 | generate_hd_keypair("localhost:10003", _SEED_A, "bot-42") |
| 224 | key = load_private_key("localhost:10003", "bot-42") |
| 225 | assert key is not None |
| 226 | assert isinstance(key, Ed25519PrivateKey) |
| 227 | |
| 228 | def test_load_private_key_agent_id_not_found_returns_none( |
| 229 | self, isolated_keys: pathlib.Path |
| 230 | ) -> None: |
| 231 | # Human key exists but no agent key |
| 232 | generate_hd_keypair("localhost:10003", _SEED_A) |
| 233 | key = load_private_key("localhost:10003", "nonexistent-agent") |
| 234 | assert key is None |
| 235 | |
| 236 | |
| 237 | # --------------------------------------------------------------------------- |
| 238 | # resolve_signing_identity — agent key → fallback chain |
| 239 | # --------------------------------------------------------------------------- |
| 240 | |
| 241 | |
| 242 | class TestResolveSigningIdentity: |
| 243 | def test_agent_key_used_when_registered( |
| 244 | self, |
| 245 | isolated_identity: pathlib.Path, |
| 246 | isolated_keys: pathlib.Path, |
| 247 | ) -> None: |
| 248 | hub = "http://localhost:10003" |
| 249 | hostname = hostname_from_url(hub) |
| 250 | agent_id = "agentception-abc" |
| 251 | |
| 252 | # Generate two distinct keys |
| 253 | generate_hd_keypair(hostname, _SEED_A) |
| 254 | generate_hd_keypair(hostname, _SEED_B, agent_id) |
| 255 | |
| 256 | agent_key_path = str(key_path_for(hostname, agent_id)) |
| 257 | save_identity( |
| 258 | hub, |
| 259 | {"type": "agent", "handle": agent_id, "key_path": agent_key_path, "algorithm": "ed25519"}, |
| 260 | agent_id=agent_id, |
| 261 | ) |
| 262 | |
| 263 | result = resolve_signing_identity(hub, agent_id=agent_id) |
| 264 | assert result is not None |
| 265 | handle, private_key = result |
| 266 | assert handle == agent_id |
| 267 | assert isinstance(private_key, Ed25519PrivateKey) |
| 268 | |
| 269 | def test_falls_back_to_human_when_no_agent_key( |
| 270 | self, |
| 271 | isolated_identity: pathlib.Path, |
| 272 | isolated_keys: pathlib.Path, |
| 273 | ) -> None: |
| 274 | hub = "http://localhost:10003" |
| 275 | hostname = hostname_from_url(hub) |
| 276 | |
| 277 | # Only human key registered |
| 278 | generate_hd_keypair(hostname, _SEED_A) |
| 279 | human_key_path = str(key_path_for(hostname)) |
| 280 | save_identity( |
| 281 | hub, |
| 282 | {"type": "human", "handle": "gabriel", "key_path": human_key_path, "algorithm": "ed25519"}, |
| 283 | ) |
| 284 | |
| 285 | # Ask for agent signing but no agent entry → falls back to human |
| 286 | result = resolve_signing_identity(hub, agent_id="unregistered-agent") |
| 287 | assert result is not None |
| 288 | handle, _ = result |
| 289 | assert handle == "gabriel" |
| 290 | |
| 291 | def test_no_identity_returns_none( |
| 292 | self, isolated_identity: pathlib.Path |
| 293 | ) -> None: |
| 294 | assert resolve_signing_identity("http://localhost:10003") is None |
| 295 | |
| 296 | def test_human_resolve_without_agent_id( |
| 297 | self, |
| 298 | isolated_identity: pathlib.Path, |
| 299 | isolated_keys: pathlib.Path, |
| 300 | ) -> None: |
| 301 | hub = "http://localhost:10003" |
| 302 | hostname = hostname_from_url(hub) |
| 303 | generate_hd_keypair(hostname, _SEED_A) |
| 304 | save_identity( |
| 305 | hub, |
| 306 | { |
| 307 | "type": "human", |
| 308 | "handle": "gabriel", |
| 309 | "key_path": str(key_path_for(hostname)), |
| 310 | "algorithm": "ed25519", |
| 311 | }, |
| 312 | ) |
| 313 | result = resolve_signing_identity(hub) |
| 314 | assert result is not None |
| 315 | handle, _ = result |
| 316 | assert handle == "gabriel" |
| 317 | |
| 318 | |
| 319 | # --------------------------------------------------------------------------- |
| 320 | # load_private_key_from_pem |
| 321 | # --------------------------------------------------------------------------- |
| 322 | |
| 323 | |
| 324 | class TestLoadPrivateKeyFromPem: |
| 325 | def _make_pem(self) -> bytes: |
| 326 | key = Ed25519PrivateKey.generate() |
| 327 | return key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()) |
| 328 | |
| 329 | def test_valid_pem_returns_key(self) -> None: |
| 330 | pem = self._make_pem() |
| 331 | key = load_private_key_from_pem(pem) |
| 332 | assert isinstance(key, Ed25519PrivateKey) |
| 333 | |
| 334 | def test_invalid_pem_returns_none(self) -> None: |
| 335 | assert load_private_key_from_pem(b"not a pem") is None |
| 336 | |
| 337 | def test_empty_bytes_returns_none(self) -> None: |
| 338 | assert load_private_key_from_pem(b"") is None |
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