test_core_hdkeys.py
python
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3
docs: docstring sprint contract→find-symbol — idiomatic run…
Sonnet 4.6
patch
142 days ago
| 1 | """Tests for muse.core.hdkeys — six-level domain-first HD key derivation. |
| 2 | |
| 3 | Test categories |
| 4 | --------------- |
| 5 | - Unit: constants, path construction, argument validation, return types |
| 6 | - Integration: full pipeline from mnemonic → seed → key → signature |
| 7 | - Stress: uniqueness across all dimensions (domain, entity, role, index) |
| 8 | - Security: domain isolation, agent sub-seed independence, least privilege |
| 9 | - Data integrity: consistency between hdkeys API and direct slip010 derivation |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import pytest |
| 15 | |
| 16 | from muse.core.bip39 import mnemonic_to_seed |
| 17 | from muse.core.hdkeys import ( |
| 18 | DOMAIN_BLOCKCHAIN, |
| 19 | DOMAIN_CODE, |
| 20 | DOMAIN_IDENTITY, |
| 21 | DOMAIN_MIDI, |
| 22 | DOMAIN_MUSIC, |
| 23 | DOMAIN_PAYMENTS, |
| 24 | DOMAIN_PROSE, |
| 25 | ENTITY_AGENT, |
| 26 | ENTITY_HUMAN, |
| 27 | ENTITY_ORG, |
| 28 | ENTITY_SERVICE, |
| 29 | HdKeyError, |
| 30 | ROLE_ATTEST, |
| 31 | ROLE_DELEGATE, |
| 32 | ROLE_PROVISION, |
| 33 | ROLE_RECEIVE, |
| 34 | ROLE_SIGN, |
| 35 | derive_agent_sub_seed, |
| 36 | derive_domain_key, |
| 37 | derive_identity_key, |
| 38 | derive_key, |
| 39 | dk_to_ed25519, |
| 40 | muse_path, |
| 41 | public_bytes_from_seed, |
| 42 | ) |
| 43 | from muse.core.slip010 import ( |
| 44 | MUSE_PURPOSE, |
| 45 | DerivedKey, |
| 46 | child_key, |
| 47 | derive_path, |
| 48 | hardened, |
| 49 | master_key, |
| 50 | ) |
| 51 | |
| 52 | _TEST_MNEMONIC = ( |
| 53 | "abandon abandon abandon abandon abandon abandon " |
| 54 | "abandon abandon abandon abandon abandon about" |
| 55 | ) |
| 56 | |
| 57 | |
| 58 | @pytest.fixture |
| 59 | def seed() -> bytes: |
| 60 | return mnemonic_to_seed(_TEST_MNEMONIC) |
| 61 | |
| 62 | |
| 63 | @pytest.fixture |
| 64 | def raw_seed() -> bytes: |
| 65 | return bytes(range(64)) |
| 66 | |
| 67 | |
| 68 | # --------------------------------------------------------------------------- |
| 69 | # Unit — constants |
| 70 | # --------------------------------------------------------------------------- |
| 71 | |
| 72 | |
| 73 | class TestConstants: |
| 74 | def test_domain_values(self) -> None: |
| 75 | assert DOMAIN_IDENTITY == 0 |
| 76 | assert DOMAIN_PAYMENTS == 1 |
| 77 | assert DOMAIN_CODE == 2 |
| 78 | assert DOMAIN_MUSIC == 3 |
| 79 | assert DOMAIN_MIDI == 4 |
| 80 | assert DOMAIN_PROSE == 5 |
| 81 | assert DOMAIN_BLOCKCHAIN == 6 |
| 82 | |
| 83 | def test_entity_values(self) -> None: |
| 84 | assert ENTITY_HUMAN == 0 |
| 85 | assert ENTITY_AGENT == 1 |
| 86 | assert ENTITY_SERVICE == 2 |
| 87 | assert ENTITY_ORG == 3 |
| 88 | |
| 89 | def test_role_values(self) -> None: |
| 90 | assert ROLE_SIGN == 0 |
| 91 | assert ROLE_RECEIVE == 1 |
| 92 | assert ROLE_PROVISION == 2 |
| 93 | assert ROLE_ATTEST == 3 |
| 94 | assert ROLE_DELEGATE == 4 |
| 95 | |
| 96 | def test_all_domains_distinct(self) -> None: |
| 97 | domains = [DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, |
| 98 | DOMAIN_MUSIC, DOMAIN_MIDI, DOMAIN_PROSE, DOMAIN_BLOCKCHAIN] |
| 99 | assert len(set(domains)) == 7 |
| 100 | |
| 101 | def test_all_entities_distinct(self) -> None: |
| 102 | entities = [ENTITY_HUMAN, ENTITY_AGENT, ENTITY_SERVICE, ENTITY_ORG] |
| 103 | assert len(set(entities)) == 4 |
| 104 | |
| 105 | def test_all_roles_distinct(self) -> None: |
| 106 | roles = [ROLE_SIGN, ROLE_RECEIVE, ROLE_PROVISION, ROLE_ATTEST, ROLE_DELEGATE] |
| 107 | assert len(set(roles)) == 5 |
| 108 | |
| 109 | |
| 110 | # --------------------------------------------------------------------------- |
| 111 | # Unit — muse_path() |
| 112 | # --------------------------------------------------------------------------- |
| 113 | |
| 114 | |
| 115 | class TestMusePath: |
| 116 | def test_default_identity_path(self) -> None: |
| 117 | assert muse_path(DOMAIN_IDENTITY) == f"m/{MUSE_PURPOSE}'/0'/0'/0'/0'/0'" |
| 118 | |
| 119 | def test_music_agent_path(self) -> None: |
| 120 | expected = f"m/{MUSE_PURPOSE}'/3'/1'/2'/0'/0'" |
| 121 | assert muse_path(DOMAIN_MUSIC, entity_type=ENTITY_AGENT, entity_id=2) == expected |
| 122 | |
| 123 | def test_code_domain_rotation(self) -> None: |
| 124 | expected = f"m/{MUSE_PURPOSE}'/2'/0'/0'/0'/1'" |
| 125 | assert muse_path(DOMAIN_CODE, index=1) == expected |
| 126 | |
| 127 | def test_payments_receive_role(self) -> None: |
| 128 | expected = f"m/{MUSE_PURPOSE}'/1'/0'/0'/1'/0'" |
| 129 | assert muse_path(DOMAIN_PAYMENTS, role=ROLE_RECEIVE) == expected |
| 130 | |
| 131 | def test_all_six_levels_present(self) -> None: |
| 132 | path = muse_path(DOMAIN_IDENTITY) |
| 133 | parts = path.split("/") |
| 134 | assert parts[0] == "m" |
| 135 | assert len(parts) == 7 # m + 6 levels |
| 136 | |
| 137 | def test_all_levels_hardened(self) -> None: |
| 138 | path = muse_path(DOMAIN_IDENTITY) |
| 139 | for part in path.split("/")[1:]: |
| 140 | assert part.endswith("'"), f"Level {part!r} is not hardened" |
| 141 | |
| 142 | def test_negative_domain_raises(self) -> None: |
| 143 | with pytest.raises(HdKeyError, match="domain"): |
| 144 | muse_path(-1) |
| 145 | |
| 146 | def test_negative_entity_id_raises(self) -> None: |
| 147 | with pytest.raises(HdKeyError, match="entity_id"): |
| 148 | muse_path(DOMAIN_IDENTITY, entity_id=-1) |
| 149 | |
| 150 | def test_negative_index_raises(self) -> None: |
| 151 | with pytest.raises(HdKeyError, match="index"): |
| 152 | muse_path(DOMAIN_IDENTITY, index=-1) |
| 153 | |
| 154 | def test_invalid_entity_type_raises(self) -> None: |
| 155 | with pytest.raises(HdKeyError, match="entity_type"): |
| 156 | muse_path(DOMAIN_IDENTITY, entity_type=4) |
| 157 | |
| 158 | def test_invalid_role_raises(self) -> None: |
| 159 | with pytest.raises(HdKeyError, match="role"): |
| 160 | muse_path(DOMAIN_IDENTITY, role=5) |
| 161 | |
| 162 | def test_returns_string(self) -> None: |
| 163 | assert isinstance(muse_path(DOMAIN_IDENTITY), str) |
| 164 | |
| 165 | def test_purpose_embedded_in_path(self) -> None: |
| 166 | assert str(MUSE_PURPOSE) in muse_path(DOMAIN_IDENTITY) |
| 167 | |
| 168 | |
| 169 | # --------------------------------------------------------------------------- |
| 170 | # Unit — derive_key() |
| 171 | # --------------------------------------------------------------------------- |
| 172 | |
| 173 | |
| 174 | class TestDeriveKey: |
| 175 | def test_returns_derived_key(self, raw_seed: bytes) -> None: |
| 176 | assert isinstance(derive_key(raw_seed, DOMAIN_IDENTITY), DerivedKey) |
| 177 | |
| 178 | def test_private_bytes_32(self, raw_seed: bytes) -> None: |
| 179 | assert len(derive_key(raw_seed, DOMAIN_IDENTITY).private_bytes) == 32 |
| 180 | |
| 181 | def test_chain_code_32(self, raw_seed: bytes) -> None: |
| 182 | assert len(derive_key(raw_seed, DOMAIN_IDENTITY).chain_code) == 32 |
| 183 | |
| 184 | def test_matches_direct_slip010_derivation(self, raw_seed: bytes) -> None: |
| 185 | path = muse_path(DOMAIN_IDENTITY, ENTITY_HUMAN, 0, ROLE_SIGN, 0) |
| 186 | dk_direct = derive_path(raw_seed, path) |
| 187 | dk_api = derive_key(raw_seed, DOMAIN_IDENTITY) |
| 188 | assert dk_api == dk_direct |
| 189 | |
| 190 | def test_negative_domain_raises(self, raw_seed: bytes) -> None: |
| 191 | with pytest.raises(HdKeyError): |
| 192 | derive_key(raw_seed, -1) |
| 193 | |
| 194 | def test_invalid_role_raises(self, raw_seed: bytes) -> None: |
| 195 | with pytest.raises(HdKeyError): |
| 196 | derive_key(raw_seed, DOMAIN_IDENTITY, role=9) |
| 197 | |
| 198 | def test_deterministic(self, raw_seed: bytes) -> None: |
| 199 | dk1 = derive_key(raw_seed, DOMAIN_MUSIC) |
| 200 | dk2 = derive_key(raw_seed, DOMAIN_MUSIC) |
| 201 | assert dk1 == dk2 |
| 202 | |
| 203 | |
| 204 | # --------------------------------------------------------------------------- |
| 205 | # Unit — derive_identity_key() |
| 206 | # --------------------------------------------------------------------------- |
| 207 | |
| 208 | |
| 209 | class TestDeriveIdentityKey: |
| 210 | def test_defaults_to_domain_identity(self, raw_seed: bytes) -> None: |
| 211 | dk = derive_identity_key(raw_seed) |
| 212 | expected = derive_path(raw_seed, muse_path(DOMAIN_IDENTITY)) |
| 213 | assert dk == expected |
| 214 | |
| 215 | def test_entity_type_agent(self, raw_seed: bytes) -> None: |
| 216 | dk = derive_identity_key(raw_seed, entity_type=ENTITY_AGENT, entity_id=0) |
| 217 | expected = derive_path(raw_seed, muse_path(DOMAIN_IDENTITY, ENTITY_AGENT, 0)) |
| 218 | assert dk == expected |
| 219 | |
| 220 | def test_rotation_index(self, raw_seed: bytes) -> None: |
| 221 | current = derive_identity_key(raw_seed, index=0) |
| 222 | rotated = derive_identity_key(raw_seed, index=1) |
| 223 | assert current != rotated |
| 224 | |
| 225 | def test_negative_entity_id_raises(self, raw_seed: bytes) -> None: |
| 226 | with pytest.raises(HdKeyError): |
| 227 | derive_identity_key(raw_seed, entity_id=-1) |
| 228 | |
| 229 | |
| 230 | # --------------------------------------------------------------------------- |
| 231 | # Unit — derive_domain_key() |
| 232 | # --------------------------------------------------------------------------- |
| 233 | |
| 234 | |
| 235 | class TestDeriveDomainKey: |
| 236 | @pytest.mark.parametrize("domain", [ |
| 237 | DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, |
| 238 | DOMAIN_MUSIC, DOMAIN_MIDI, DOMAIN_PROSE, DOMAIN_BLOCKCHAIN, |
| 239 | ]) |
| 240 | def test_all_named_domains(self, domain: int, raw_seed: bytes) -> None: |
| 241 | dk = derive_domain_key(raw_seed, domain) |
| 242 | assert isinstance(dk, DerivedKey) |
| 243 | |
| 244 | def test_code_differs_from_music(self, raw_seed: bytes) -> None: |
| 245 | assert derive_domain_key(raw_seed, DOMAIN_CODE) != derive_domain_key(raw_seed, DOMAIN_MUSIC) |
| 246 | |
| 247 | def test_role_receive_differs_from_sign(self, raw_seed: bytes) -> None: |
| 248 | sign = derive_domain_key(raw_seed, DOMAIN_PAYMENTS, role=ROLE_SIGN) |
| 249 | recv = derive_domain_key(raw_seed, DOMAIN_PAYMENTS, role=ROLE_RECEIVE) |
| 250 | assert sign != recv |
| 251 | |
| 252 | |
| 253 | # --------------------------------------------------------------------------- |
| 254 | # Unit — derive_agent_sub_seed() |
| 255 | # --------------------------------------------------------------------------- |
| 256 | |
| 257 | |
| 258 | class TestDeriveAgentSubSeed: |
| 259 | def test_returns_64_bytes(self, raw_seed: bytes) -> None: |
| 260 | sub = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 261 | assert isinstance(sub, (bytes, bytearray)) and len(sub) == 64 |
| 262 | |
| 263 | def test_different_domains_produce_different_sub_seeds(self, raw_seed: bytes) -> None: |
| 264 | sub_music = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 265 | sub_code = derive_agent_sub_seed(raw_seed, DOMAIN_CODE, agent_id=0) |
| 266 | assert sub_music != sub_code |
| 267 | |
| 268 | def test_different_agent_ids_produce_different_sub_seeds(self, raw_seed: bytes) -> None: |
| 269 | sub0 = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 270 | sub1 = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=1) |
| 271 | assert sub0 != sub1 |
| 272 | |
| 273 | def test_deterministic(self, raw_seed: bytes) -> None: |
| 274 | sub1 = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 275 | sub2 = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 276 | assert sub1 == sub2 |
| 277 | |
| 278 | def test_differs_from_master_seed(self, raw_seed: bytes) -> None: |
| 279 | sub = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 280 | assert sub != raw_seed |
| 281 | |
| 282 | def test_sub_seed_composition(self, raw_seed: bytes) -> None: |
| 283 | """Sub-seed = private_bytes + chain_code at m/purpose'/domain'/AGENT'/agent_id'.""" |
| 284 | dk = master_key(raw_seed) |
| 285 | dk = child_key(dk, hardened(MUSE_PURPOSE)) |
| 286 | dk = child_key(dk, hardened(DOMAIN_MUSIC)) |
| 287 | dk = child_key(dk, hardened(ENTITY_AGENT)) |
| 288 | dk = child_key(dk, hardened(0)) |
| 289 | expected = dk.private_bytes + dk.chain_code |
| 290 | assert derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) == expected |
| 291 | |
| 292 | def test_agent_can_derive_identity_key_from_sub_seed(self, raw_seed: bytes) -> None: |
| 293 | auth_seed = derive_agent_sub_seed(raw_seed, DOMAIN_IDENTITY, agent_id=0) |
| 294 | dk = derive_identity_key(auth_seed) |
| 295 | assert isinstance(dk, DerivedKey) |
| 296 | |
| 297 | def test_negative_domain_raises(self, raw_seed: bytes) -> None: |
| 298 | with pytest.raises(HdKeyError): |
| 299 | derive_agent_sub_seed(raw_seed, -1, agent_id=0) |
| 300 | |
| 301 | def test_negative_agent_id_raises(self, raw_seed: bytes) -> None: |
| 302 | with pytest.raises(HdKeyError): |
| 303 | derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=-1) |
| 304 | |
| 305 | |
| 306 | # --------------------------------------------------------------------------- |
| 307 | # Unit — dk_to_ed25519() and public_bytes_from_seed() |
| 308 | # --------------------------------------------------------------------------- |
| 309 | |
| 310 | |
| 311 | class TestKeyMaterialisation: |
| 312 | def test_dk_to_ed25519_returns_signing_key(self, raw_seed: bytes) -> None: |
| 313 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 314 | dk = derive_identity_key(raw_seed) |
| 315 | assert isinstance(dk_to_ed25519(dk), Ed25519PrivateKey) |
| 316 | |
| 317 | def test_sign_and_verify(self, raw_seed: bytes) -> None: |
| 318 | dk = derive_identity_key(raw_seed) |
| 319 | priv = dk_to_ed25519(dk) |
| 320 | msg = b"muse hd path design" |
| 321 | priv.public_key().verify(priv.sign(msg), msg) |
| 322 | |
| 323 | def test_public_bytes_32(self, raw_seed: bytes) -> None: |
| 324 | assert len(public_bytes_from_seed(raw_seed)) == 32 |
| 325 | |
| 326 | def test_public_bytes_deterministic(self, raw_seed: bytes) -> None: |
| 327 | assert public_bytes_from_seed(raw_seed) == public_bytes_from_seed(raw_seed) |
| 328 | |
| 329 | def test_public_bytes_defaults_to_identity_domain(self, raw_seed: bytes) -> None: |
| 330 | pub = public_bytes_from_seed(raw_seed) |
| 331 | expected = dk_to_ed25519(derive_identity_key(raw_seed)).public_key().public_bytes_raw() |
| 332 | assert pub == expected |
| 333 | |
| 334 | def test_different_domains_different_public_keys(self, raw_seed: bytes) -> None: |
| 335 | pub_id = public_bytes_from_seed(raw_seed, domain=DOMAIN_IDENTITY) |
| 336 | pub_code = public_bytes_from_seed(raw_seed, domain=DOMAIN_CODE) |
| 337 | assert pub_id != pub_code |
| 338 | |
| 339 | |
| 340 | # --------------------------------------------------------------------------- |
| 341 | # Integration — full pipeline |
| 342 | # --------------------------------------------------------------------------- |
| 343 | |
| 344 | |
| 345 | class TestFullPipeline: |
| 346 | def test_mnemonic_to_signature(self, seed: bytes) -> None: |
| 347 | dk = derive_identity_key(seed) |
| 348 | priv = dk_to_ed25519(dk) |
| 349 | msg = b"muse authentication" |
| 350 | priv.public_key().verify(priv.sign(msg), msg) |
| 351 | |
| 352 | def test_wrong_key_cannot_verify(self, seed: bytes) -> None: |
| 353 | from cryptography.exceptions import InvalidSignature |
| 354 | current = derive_identity_key(seed, index=0) |
| 355 | rotated = derive_identity_key(seed, index=1) |
| 356 | msg = b"signed with current key" |
| 357 | sig = dk_to_ed25519(current).sign(msg) |
| 358 | with pytest.raises(InvalidSignature): |
| 359 | dk_to_ed25519(rotated).public_key().verify(sig, msg) |
| 360 | |
| 361 | def test_human_and_agent_identity_keys_differ(self, seed: bytes) -> None: |
| 362 | auth_seed = derive_agent_sub_seed(seed, DOMAIN_IDENTITY, agent_id=0) |
| 363 | pub_human = public_bytes_from_seed(seed, domain=DOMAIN_IDENTITY) |
| 364 | pub_agent = dk_to_ed25519(derive_identity_key(auth_seed)).public_key().public_bytes_raw() |
| 365 | assert pub_human != pub_agent |
| 366 | |
| 367 | def test_five_agents_distinct_identity_keys(self, seed: bytes) -> None: |
| 368 | pub_keys: set[bytes] = set() |
| 369 | for agent_id in range(5): |
| 370 | auth_seed = derive_agent_sub_seed(seed, DOMAIN_IDENTITY, agent_id=agent_id) |
| 371 | pub = dk_to_ed25519(derive_identity_key(auth_seed)).public_key().public_bytes_raw() |
| 372 | assert pub not in pub_keys, f"Duplicate at agent_id={agent_id}" |
| 373 | pub_keys.add(pub) |
| 374 | |
| 375 | |
| 376 | # --------------------------------------------------------------------------- |
| 377 | # Security — domain isolation and least privilege |
| 378 | # --------------------------------------------------------------------------- |
| 379 | |
| 380 | |
| 381 | class TestSecurity: |
| 382 | def test_identity_key_differs_from_all_domain_keys(self, raw_seed: bytes) -> None: |
| 383 | identity_pub = public_bytes_from_seed(raw_seed, domain=DOMAIN_IDENTITY) |
| 384 | for domain in (DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC, |
| 385 | DOMAIN_MIDI, DOMAIN_PROSE, DOMAIN_BLOCKCHAIN): |
| 386 | other_pub = public_bytes_from_seed(raw_seed, domain=domain) |
| 387 | assert identity_pub != other_pub, f"Identity key leaked into domain={domain}" |
| 388 | |
| 389 | def test_music_agent_cannot_derive_code_domain_key(self, raw_seed: bytes) -> None: |
| 390 | """A music agent's sub-seed produces a different code key than the operator's.""" |
| 391 | music_agent_seed = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 392 | operator_code_pub = public_bytes_from_seed(raw_seed, domain=DOMAIN_CODE) |
| 393 | agent_code_pub = public_bytes_from_seed(music_agent_seed, domain=DOMAIN_CODE) |
| 394 | assert operator_code_pub != agent_code_pub |
| 395 | |
| 396 | def test_domain_scoped_sub_seeds_are_independent(self, raw_seed: bytes) -> None: |
| 397 | domains = [DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, |
| 398 | DOMAIN_MUSIC, DOMAIN_MIDI, DOMAIN_PROSE] |
| 399 | sub_seeds = [bytes(derive_agent_sub_seed(raw_seed, d, agent_id=0)) for d in domains] |
| 400 | assert len(set(sub_seeds)) == len(domains), "Domain sub-seeds are not independent" |
| 401 | |
| 402 | def test_all_roles_produce_distinct_keys(self, raw_seed: bytes) -> None: |
| 403 | keys: set[bytes] = set() |
| 404 | for role in (ROLE_SIGN, ROLE_RECEIVE, ROLE_PROVISION, ROLE_ATTEST, ROLE_DELEGATE): |
| 405 | pub = public_bytes_from_seed(raw_seed, domain=DOMAIN_CODE, role=role) |
| 406 | assert pub not in keys, f"Duplicate key for role={role}" |
| 407 | keys.add(pub) |
| 408 | |
| 409 | def test_sub_seed_does_not_expose_master_seed(self, raw_seed: bytes) -> None: |
| 410 | sub = derive_agent_sub_seed(raw_seed, DOMAIN_MUSIC, agent_id=0) |
| 411 | assert sub != raw_seed |
| 412 | # First 32 bytes of sub-seed are derived private bytes — must not equal |
| 413 | # anything trivially derivable from master seed |
| 414 | assert sub[:32] != raw_seed[:32] |
| 415 | |
| 416 | |
| 417 | # --------------------------------------------------------------------------- |
| 418 | # Stress — uniqueness across all six dimensions |
| 419 | # --------------------------------------------------------------------------- |
| 420 | |
| 421 | |
| 422 | class TestStress: |
| 423 | def test_100_domain_keys_all_unique(self, raw_seed: bytes) -> None: |
| 424 | # Extend domain namespace — open by design |
| 425 | pubs: set[bytes] = set() |
| 426 | for domain in range(100): |
| 427 | pub = public_bytes_from_seed(raw_seed, domain=domain) |
| 428 | assert pub not in pubs, f"Duplicate at domain={domain}" |
| 429 | pubs.add(pub) |
| 430 | |
| 431 | def test_100_entity_ids_all_unique(self, raw_seed: bytes) -> None: |
| 432 | pubs: set[bytes] = set() |
| 433 | for entity_id in range(100): |
| 434 | pub = public_bytes_from_seed(raw_seed, domain=DOMAIN_IDENTITY, entity_id=entity_id) |
| 435 | assert pub not in pubs, f"Duplicate at entity_id={entity_id}" |
| 436 | pubs.add(pub) |
| 437 | |
| 438 | def test_100_rotation_indices_all_unique(self, raw_seed: bytes) -> None: |
| 439 | pubs: set[bytes] = set() |
| 440 | for index in range(100): |
| 441 | pub = public_bytes_from_seed(raw_seed, domain=DOMAIN_IDENTITY, index=index) |
| 442 | assert pub not in pubs, f"Duplicate at index={index}" |
| 443 | pubs.add(pub) |
| 444 | |
| 445 | def test_50_agent_sub_seeds_all_unique_per_domain(self, raw_seed: bytes) -> None: |
| 446 | for domain in (DOMAIN_IDENTITY, DOMAIN_MUSIC, DOMAIN_CODE): |
| 447 | subs: set[bytes] = set() |
| 448 | for agent_id in range(50): |
| 449 | sub = bytes(derive_agent_sub_seed(raw_seed, domain, agent_id=agent_id)) |
| 450 | assert sub not in subs, f"Duplicate at domain={domain} agent_id={agent_id}" |
| 451 | subs.add(sub) |
| 452 | |
| 453 | def test_all_7_domains_x_4_entities_x_5_roles_unique(self, raw_seed: bytes) -> None: |
| 454 | domains = [DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC, |
| 455 | DOMAIN_MIDI, DOMAIN_PROSE, DOMAIN_BLOCKCHAIN] |
| 456 | entities = [ENTITY_HUMAN, ENTITY_AGENT, ENTITY_SERVICE, ENTITY_ORG] |
| 457 | roles = [ROLE_SIGN, ROLE_RECEIVE, ROLE_PROVISION, ROLE_ATTEST, ROLE_DELEGATE] |
| 458 | pubs: set[bytes] = set() |
| 459 | total = 0 |
| 460 | for d in domains: |
| 461 | for e in entities: |
| 462 | for r in roles: |
| 463 | pub = public_bytes_from_seed(raw_seed, domain=d, entity_type=e, role=r) |
| 464 | assert pub not in pubs, f"Duplicate at domain={d} entity={e} role={r}" |
| 465 | pubs.add(pub) |
| 466 | total += 1 |
| 467 | assert total == 7 * 4 * 5 # 140 unique keys |
| 468 | |
| 469 | |
| 470 | # --------------------------------------------------------------------------- |
| 471 | # End-to-end |
| 472 | # --------------------------------------------------------------------------- |
| 473 | |
| 474 | |
| 475 | class TestEndToEnd: |
| 476 | """Full workflow scenarios from mnemonic to network-ready public key. |
| 477 | |
| 478 | These tests simulate real caller sequences: the path a MuseHub |
| 479 | registration, an agent spawn, and a key rotation take through this stack. |
| 480 | """ |
| 481 | |
| 482 | def test_musehub_registration_flow(self) -> None: |
| 483 | """Operator generates mnemonic, derives identity key, gets public key for registration.""" |
| 484 | from muse.core.bip39 import generate_mnemonic, mnemonic_to_seed |
| 485 | mnemonic = generate_mnemonic() |
| 486 | seed = mnemonic_to_seed(mnemonic) |
| 487 | pub = public_bytes_from_seed(seed, domain=DOMAIN_IDENTITY) |
| 488 | assert len(pub) == 32 |
| 489 | fingerprint = pub.hex() |
| 490 | assert len(fingerprint) == 64 # 32 bytes → 64 hex chars |
| 491 | |
| 492 | def test_agent_spawn_flow(self) -> None: |
| 493 | """Operator spawns a music agent with domain-scoped sub-seed.""" |
| 494 | from muse.core.bip39 import generate_mnemonic, mnemonic_to_seed |
| 495 | seed = mnemonic_to_seed(generate_mnemonic()) |
| 496 | |
| 497 | # Operator derives domain-scoped sub-seeds for the agent |
| 498 | music_seed = derive_agent_sub_seed(seed, DOMAIN_MUSIC, agent_id=0) |
| 499 | auth_seed = derive_agent_sub_seed(seed, DOMAIN_IDENTITY, agent_id=0) |
| 500 | |
| 501 | # Agent uses each seed for its respective domain |
| 502 | agent_identity_pub = dk_to_ed25519(derive_identity_key(auth_seed)).public_key().public_bytes_raw() |
| 503 | agent_music_pub = dk_to_ed25519(derive_domain_key(music_seed, DOMAIN_MUSIC)).public_key().public_bytes_raw() |
| 504 | |
| 505 | assert len(agent_identity_pub) == 32 |
| 506 | assert len(agent_music_pub) == 32 |
| 507 | # Agent's identity pub differs from operator's |
| 508 | operator_pub = public_bytes_from_seed(seed, domain=DOMAIN_IDENTITY) |
| 509 | assert agent_identity_pub != operator_pub |
| 510 | |
| 511 | def test_key_rotation_flow(self) -> None: |
| 512 | """Operator pre-registers next key then rotates.""" |
| 513 | from muse.core.bip39 import generate_mnemonic, mnemonic_to_seed |
| 514 | seed = mnemonic_to_seed(generate_mnemonic()) |
| 515 | |
| 516 | # Current key in use |
| 517 | current_pub = public_bytes_from_seed(seed, domain=DOMAIN_IDENTITY, index=0) |
| 518 | # Pre-register next key before rotation |
| 519 | next_pub = public_bytes_from_seed(seed, domain=DOMAIN_IDENTITY, index=1) |
| 520 | |
| 521 | assert current_pub != next_pub |
| 522 | # After rotation, index 1 becomes the active key |
| 523 | # Both remain independently derivable from the same seed |
| 524 | assert len(current_pub) == len(next_pub) == 32 |
| 525 | |
| 526 | def test_commit_signing_flow(self) -> None: |
| 527 | """Operator signs a commit hash with their code domain key.""" |
| 528 | from muse.core.bip39 import generate_mnemonic, mnemonic_to_seed |
| 529 | import hashlib |
| 530 | seed = mnemonic_to_seed(generate_mnemonic()) |
| 531 | dk = derive_domain_key(seed, domain=DOMAIN_CODE) |
| 532 | priv = dk_to_ed25519(dk) |
| 533 | # Simulate a commit hash |
| 534 | commit_bytes = hashlib.sha256(b"tree abc123\nparent def456\nauthor gabriel").digest() |
| 535 | sig = priv.sign(commit_bytes) |
| 536 | # Any verifier with the public key can verify |
| 537 | priv.public_key().verify(sig, commit_bytes) |
| 538 | |
| 539 | def test_music_project_signing_flow(self) -> None: |
| 540 | """Operator signs a Stori project snapshot with their music domain key.""" |
| 541 | from muse.core.bip39 import generate_mnemonic, mnemonic_to_seed |
| 542 | seed = mnemonic_to_seed(generate_mnemonic()) |
| 543 | dk = derive_domain_key(seed, domain=DOMAIN_MUSIC) |
| 544 | priv = dk_to_ed25519(dk) |
| 545 | project_manifest = b"track: neon_sunrise.wav\nbpm: 128\nkey: Am" |
| 546 | sig = priv.sign(project_manifest) |
| 547 | priv.public_key().verify(sig, project_manifest) |
| 548 | |
| 549 | |
| 550 | # --------------------------------------------------------------------------- |
| 551 | # Performance |
| 552 | # --------------------------------------------------------------------------- |
| 553 | |
| 554 | |
| 555 | class TestPerformance: |
| 556 | """Timing budgets for hdkeys high-level operations. |
| 557 | |
| 558 | Key derivation is dominated by HMAC-SHA512 (fast). The only slow |
| 559 | operation is mnemonic_to_seed (PBKDF2) which lives in bip39 — once |
| 560 | the seed is in hand, all hdkeys operations must be fast. |
| 561 | """ |
| 562 | |
| 563 | def test_derive_identity_key_under_5ms(self) -> None: |
| 564 | import time |
| 565 | seed = bytes(range(64)) |
| 566 | start = time.perf_counter() |
| 567 | for _ in range(100): |
| 568 | derive_identity_key(seed) |
| 569 | elapsed = (time.perf_counter() - start) / 100 |
| 570 | assert elapsed < 0.005, f"derive_identity_key averaged {elapsed*1000:.2f}ms — too slow" |
| 571 | |
| 572 | def test_derive_domain_key_under_5ms(self) -> None: |
| 573 | import time |
| 574 | seed = bytes(range(64)) |
| 575 | start = time.perf_counter() |
| 576 | for _ in range(100): |
| 577 | derive_domain_key(seed, DOMAIN_MUSIC) |
| 578 | elapsed = (time.perf_counter() - start) / 100 |
| 579 | assert elapsed < 0.005, f"derive_domain_key averaged {elapsed*1000:.2f}ms — too slow" |
| 580 | |
| 581 | def test_derive_agent_sub_seed_under_5ms(self) -> None: |
| 582 | import time |
| 583 | seed = bytes(range(64)) |
| 584 | start = time.perf_counter() |
| 585 | for _ in range(100): |
| 586 | derive_agent_sub_seed(seed, DOMAIN_MUSIC, agent_id=0) |
| 587 | elapsed = (time.perf_counter() - start) / 100 |
| 588 | assert elapsed < 0.005, f"derive_agent_sub_seed averaged {elapsed*1000:.2f}ms — too slow" |
| 589 | |
| 590 | def test_public_bytes_from_seed_under_5ms(self) -> None: |
| 591 | import time |
| 592 | seed = bytes(range(64)) |
| 593 | start = time.perf_counter() |
| 594 | for _ in range(100): |
| 595 | public_bytes_from_seed(seed) |
| 596 | elapsed = (time.perf_counter() - start) / 100 |
| 597 | assert elapsed < 0.005, f"public_bytes_from_seed averaged {elapsed*1000:.2f}ms — too slow" |
| 598 | |
| 599 | def test_140_key_matrix_under_100ms(self) -> None: |
| 600 | """7 domains × 4 entity types × 5 roles = 140 derivations must complete in < 100ms.""" |
| 601 | import time |
| 602 | seed = bytes(range(64)) |
| 603 | domains = [DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC, |
| 604 | DOMAIN_MIDI, DOMAIN_PROSE, DOMAIN_BLOCKCHAIN] |
| 605 | entities = [ENTITY_HUMAN, ENTITY_AGENT, ENTITY_SERVICE, ENTITY_ORG] |
| 606 | roles = [ROLE_SIGN, ROLE_RECEIVE, ROLE_PROVISION, ROLE_ATTEST, ROLE_DELEGATE] |
| 607 | start = time.perf_counter() |
| 608 | for d in domains: |
| 609 | for e in entities: |
| 610 | for r in roles: |
| 611 | derive_key(seed, domain=d, entity_type=e, role=r) |
| 612 | elapsed = time.perf_counter() - start |
| 613 | assert elapsed < 0.1, f"140-key matrix took {elapsed*1000:.1f}ms — too slow" |
| 614 | |
| 615 | |
| 616 | # --------------------------------------------------------------------------- |
| 617 | # Docstrings |
| 618 | # --------------------------------------------------------------------------- |
| 619 | |
| 620 | |
| 621 | class TestDocstrings: |
| 622 | """Every public symbol in muse.core.hdkeys must have a docstring.""" |
| 623 | |
| 624 | def test_module_has_docstring(self) -> None: |
| 625 | import muse.core.hdkeys as mod |
| 626 | assert mod.__doc__, "muse.core.hdkeys module has no docstring" |
| 627 | |
| 628 | @pytest.mark.parametrize("name", [ |
| 629 | "HdKeyError", |
| 630 | "muse_path", |
| 631 | "derive_key", |
| 632 | "derive_identity_key", |
| 633 | "derive_domain_key", |
| 634 | "derive_agent_sub_seed", |
| 635 | "dk_to_ed25519", |
| 636 | "public_bytes_from_seed", |
| 637 | ]) |
| 638 | def test_public_symbol_has_docstring(self, name: str) -> None: |
| 639 | import muse.core.hdkeys as mod |
| 640 | obj = getattr(mod, name) |
| 641 | assert obj.__doc__, f"muse.core.hdkeys.{name} has no docstring" |
File History
2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3
docs: docstring sprint contract→find-symbol — idiomatic run…
Sonnet 4.6
patch
142 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9
fix(cursorignore): remove git-ism (.git/worktrees)
Human
minor
⚠
145 days ago