test_provenance.py
python
| 1 | """Tests for muse.core.provenance — AgentIdentity, HMAC signing, key I/O.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import pathlib |
| 5 | import tempfile |
| 6 | |
| 7 | import pytest |
| 8 | |
| 9 | from muse.core.provenance import ( |
| 10 | AgentIdentity, |
| 11 | generate_agent_key, |
| 12 | key_fingerprint, |
| 13 | make_agent_identity, |
| 14 | read_agent_key, |
| 15 | sign_commit_hmac, |
| 16 | sign_commit_record, |
| 17 | verify_commit_hmac, |
| 18 | write_agent_key, |
| 19 | ) |
| 20 | from muse.core.store import CommitRecord |
| 21 | import datetime |
| 22 | |
| 23 | |
| 24 | # --------------------------------------------------------------------------- |
| 25 | # AgentIdentity factory |
| 26 | # --------------------------------------------------------------------------- |
| 27 | |
| 28 | |
| 29 | class TestMakeAgentIdentity: |
| 30 | def test_required_fields_present(self) -> None: |
| 31 | identity = make_agent_identity( |
| 32 | agent_id="test-agent", |
| 33 | model_id="gpt-5", |
| 34 | toolchain_id="muse-v2", |
| 35 | ) |
| 36 | assert identity["agent_id"] == "test-agent" |
| 37 | assert identity.get("model_id") == "gpt-5" |
| 38 | assert identity.get("toolchain_id") == "muse-v2" |
| 39 | |
| 40 | def test_prompt_hash_is_hex(self) -> None: |
| 41 | identity = make_agent_identity( |
| 42 | agent_id="a", |
| 43 | model_id="m", |
| 44 | toolchain_id="t", |
| 45 | prompt="system: you are a music agent", |
| 46 | ) |
| 47 | prompt_hash = identity.get("prompt_hash", "") |
| 48 | assert isinstance(prompt_hash, str) |
| 49 | assert len(prompt_hash) == 64 |
| 50 | assert all(c in "0123456789abcdef" for c in prompt_hash) |
| 51 | |
| 52 | def test_no_prompt_gives_no_hash_key(self) -> None: |
| 53 | identity = make_agent_identity(agent_id="a", model_id="m", toolchain_id="t") |
| 54 | # When no prompt is provided, prompt_hash is absent (total=False TypedDict). |
| 55 | assert identity.get("prompt_hash", "") == "" |
| 56 | |
| 57 | def test_execution_context_hash_populated(self) -> None: |
| 58 | identity = make_agent_identity( |
| 59 | agent_id="a", |
| 60 | model_id="m", |
| 61 | toolchain_id="t", |
| 62 | execution_context='{"env": "ci", "version": "1.2.3"}', |
| 63 | ) |
| 64 | ec_hash = identity.get("execution_context_hash", "") |
| 65 | assert isinstance(ec_hash, str) |
| 66 | assert len(ec_hash) == 64 |
| 67 | |
| 68 | |
| 69 | # --------------------------------------------------------------------------- |
| 70 | # Key generation and fingerprinting |
| 71 | # --------------------------------------------------------------------------- |
| 72 | |
| 73 | |
| 74 | class TestKeyGeneration: |
| 75 | def test_generate_key_is_32_bytes(self) -> None: |
| 76 | key = generate_agent_key() |
| 77 | assert isinstance(key, bytes) |
| 78 | assert len(key) == 32 |
| 79 | |
| 80 | def test_keys_are_unique(self) -> None: |
| 81 | keys = {generate_agent_key() for _ in range(10)} |
| 82 | assert len(keys) == 10 |
| 83 | |
| 84 | def test_fingerprint_is_short_hex(self) -> None: |
| 85 | key = generate_agent_key() |
| 86 | fp = key_fingerprint(key) |
| 87 | assert isinstance(fp, str) |
| 88 | assert 8 <= len(fp) <= 16 |
| 89 | assert all(c in "0123456789abcdef" for c in fp) |
| 90 | |
| 91 | def test_fingerprint_is_deterministic(self) -> None: |
| 92 | key = b"\x01" * 32 |
| 93 | assert key_fingerprint(key) == key_fingerprint(key) |
| 94 | |
| 95 | |
| 96 | # --------------------------------------------------------------------------- |
| 97 | # Key I/O |
| 98 | # --------------------------------------------------------------------------- |
| 99 | |
| 100 | |
| 101 | class TestKeyIO: |
| 102 | def test_write_and_read_roundtrip(self, tmp_path: pathlib.Path) -> None: |
| 103 | key = generate_agent_key() |
| 104 | agent_id = "roundtrip-agent" |
| 105 | write_agent_key(tmp_path, agent_id, key) |
| 106 | recovered = read_agent_key(tmp_path, agent_id) |
| 107 | assert recovered == key |
| 108 | |
| 109 | def test_read_missing_key_returns_none(self, tmp_path: pathlib.Path) -> None: |
| 110 | result = read_agent_key(tmp_path, "nonexistent-agent") |
| 111 | assert result is None |
| 112 | |
| 113 | def test_key_file_is_readable(self, tmp_path: pathlib.Path) -> None: |
| 114 | key = b"\xde\xad\xbe\xef" + b"\x00" * 28 |
| 115 | write_agent_key(tmp_path, "hex-agent", key) |
| 116 | # Find the key file and verify it roundtrips correctly. |
| 117 | key_dir = tmp_path / ".muse" / "keys" |
| 118 | files = list(key_dir.rglob("*.key")) |
| 119 | assert files |
| 120 | recovered = read_agent_key(tmp_path, "hex-agent") |
| 121 | assert recovered == key |
| 122 | |
| 123 | |
| 124 | # --------------------------------------------------------------------------- |
| 125 | # HMAC signing / verification |
| 126 | # --------------------------------------------------------------------------- |
| 127 | |
| 128 | |
| 129 | class TestHMACSigning: |
| 130 | def test_sign_and_verify_succeed(self) -> None: |
| 131 | key = generate_agent_key() |
| 132 | commit_hash = "abc123def456" * 4 |
| 133 | sig = sign_commit_hmac(commit_hash, key) |
| 134 | assert verify_commit_hmac(commit_hash, sig, key) |
| 135 | |
| 136 | def test_wrong_key_fails(self) -> None: |
| 137 | key1 = generate_agent_key() |
| 138 | key2 = generate_agent_key() |
| 139 | commit_hash = "abc123" |
| 140 | sig = sign_commit_hmac(commit_hash, key1) |
| 141 | assert not verify_commit_hmac(commit_hash, sig, key2) |
| 142 | |
| 143 | def test_wrong_commit_hash_fails(self) -> None: |
| 144 | key = generate_agent_key() |
| 145 | sig = sign_commit_hmac("commit-a", key) |
| 146 | assert not verify_commit_hmac("commit-b", sig, key) |
| 147 | |
| 148 | def test_tampered_signature_fails(self) -> None: |
| 149 | key = generate_agent_key() |
| 150 | sig = sign_commit_hmac("abc", key) |
| 151 | tampered = sig[:-4] + "0000" |
| 152 | assert not verify_commit_hmac("abc", tampered, key) |
| 153 | |
| 154 | def test_signature_is_hex_string(self) -> None: |
| 155 | key = generate_agent_key() |
| 156 | sig = sign_commit_hmac("test-commit", key) |
| 157 | assert isinstance(sig, str) |
| 158 | assert all(c in "0123456789abcdef" for c in sig) |
| 159 | |
| 160 | def test_signature_length_is_64(self) -> None: |
| 161 | key = generate_agent_key() |
| 162 | sig = sign_commit_hmac("test-commit", key) |
| 163 | assert len(sig) == 64 # HMAC-SHA256 produces 32 bytes = 64 hex chars |
| 164 | |
| 165 | |
| 166 | # --------------------------------------------------------------------------- |
| 167 | # sign_commit_record |
| 168 | # --------------------------------------------------------------------------- |
| 169 | |
| 170 | |
| 171 | class TestSignCommitRecord: |
| 172 | def test_sign_commit_record_writes_signature(self, tmp_path: pathlib.Path) -> None: |
| 173 | key = generate_agent_key() |
| 174 | agent_id = "sign-test-agent" |
| 175 | write_agent_key(tmp_path, agent_id, key) |
| 176 | |
| 177 | commit_id = "deadbeef" * 8 |
| 178 | result = sign_commit_record(commit_id, agent_id, tmp_path) |
| 179 | assert result is not None |
| 180 | sig, fprint = result |
| 181 | assert sig != "" |
| 182 | assert fprint == key_fingerprint(key) |
| 183 | assert verify_commit_hmac(commit_id, sig, key) |
| 184 | |
| 185 | def test_sign_commit_record_no_key_returns_none(self, tmp_path: pathlib.Path) -> None: |
| 186 | result = sign_commit_record("aabbccdd" * 8, "ghost-agent", tmp_path) |
| 187 | assert result is None |