test_knowtation_quorum.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Tests for Phase 7.7 — org-quorum signing via the identity domain. |
| 2 | |
| 3 | Coverage tiers (Rule #0) |
| 4 | ------------------------ |
| 5 | |
| 6 | * unit — handle validation, identity-record shape coercion, |
| 7 | filename encoding, build_quorum_metadata edge cases, |
| 8 | quorum-math threshold edges, identity-chain cycle detection. |
| 9 | * integration — KnowtationFsIdentityProvider round-trip (write JSON → |
| 10 | read back), 3-member 2-of-3 quorum with all permutations |
| 11 | of two valid signers verifying. |
| 12 | * security — partial-quorum forgery rejected; signer collusion below |
| 13 | threshold rejected; signature stripping detected at |
| 14 | commit time; signer public_key mismatch rejected; |
| 15 | cycle-introducing membership detected; depth-bounded |
| 16 | recursion; revoked members do not count; path-traversal |
| 17 | handles rejected; provider missing → partial=True (never |
| 18 | silent True). |
| 19 | * end-to-end — N/A (covered by the CLI integration in |
| 20 | test_cli_workflow once available; this suite stays focused |
| 21 | on the engine + provider). |
| 22 | """ |
| 23 | |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import base64 |
| 27 | import datetime |
| 28 | import json |
| 29 | import pathlib |
| 30 | import threading |
| 31 | from typing import Any |
| 32 | |
| 33 | import pytest |
| 34 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 35 | from cryptography.hazmat.primitives.serialization import ( |
| 36 | Encoding, NoEncryption, PrivateFormat, PublicFormat, |
| 37 | ) |
| 38 | |
| 39 | from muse.core.attestation import ( |
| 40 | AttestationVerifyError, |
| 41 | IdentityRecord, |
| 42 | register_identity_provider, |
| 43 | unregister_identity_provider, |
| 44 | ) |
| 45 | from muse.core.provenance import ( |
| 46 | encode_public_key, |
| 47 | provenance_payload, |
| 48 | sign_commit_ed25519, |
| 49 | ) |
| 50 | from muse.core.verify import ( |
| 51 | DEFAULT_IDENTITY_DEPTH_LIMIT, |
| 52 | QuorumVerifyResult, |
| 53 | verify_identity_chain, |
| 54 | verify_quorum_signatures, |
| 55 | ) |
| 56 | from muse.plugins.knowtation.quorum import ( |
| 57 | KnowtationFsIdentityProvider, |
| 58 | PROVIDER_DOMAIN, |
| 59 | build_quorum_metadata, |
| 60 | is_valid_handle, |
| 61 | register_knowtation_identity_provider, |
| 62 | ) |
| 63 | |
| 64 | |
| 65 | # --------------------------------------------------------------------------- |
| 66 | # Test helpers |
| 67 | # --------------------------------------------------------------------------- |
| 68 | |
| 69 | |
| 70 | def _new_keypair() -> tuple[Ed25519PrivateKey, str]: |
| 71 | """Return ``(private_key, base64url_public_key_no_padding)``.""" |
| 72 | priv = Ed25519PrivateKey.generate() |
| 73 | _raw, pub_b64 = encode_public_key(priv) |
| 74 | return priv, pub_b64 |
| 75 | |
| 76 | |
| 77 | def _write_identity( |
| 78 | dir_: pathlib.Path, |
| 79 | handle: str, |
| 80 | *, |
| 81 | kind: str, |
| 82 | public_key: str = "", |
| 83 | members: list[str] | None = None, |
| 84 | quorum_threshold: int = 0, |
| 85 | revoked_at: str = "", |
| 86 | ) -> None: |
| 87 | """Persist an identity record JSON file under *dir_*.""" |
| 88 | dir_.mkdir(parents=True, exist_ok=True) |
| 89 | fname = ("_at_" + handle[1:]) if handle.startswith("@") else handle |
| 90 | body: dict[str, Any] = { |
| 91 | "handle": handle, |
| 92 | "kind": kind, |
| 93 | "public_key": public_key, |
| 94 | } |
| 95 | if members is not None: |
| 96 | body["members"] = members |
| 97 | if quorum_threshold: |
| 98 | body["quorum_threshold"] = quorum_threshold |
| 99 | if revoked_at: |
| 100 | body["revoked_at"] = revoked_at |
| 101 | (dir_ / f"{fname}.json").write_text(json.dumps(body), encoding="utf-8") |
| 102 | |
| 103 | |
| 104 | class _FakeCommit: |
| 105 | """Minimal CommitRecord-shaped duck for verify_quorum_signatures.""" |
| 106 | |
| 107 | def __init__( |
| 108 | self, |
| 109 | commit_id: str, |
| 110 | metadata: dict[str, Any], |
| 111 | *, |
| 112 | author: str = "", |
| 113 | agent_id: str = "", |
| 114 | committed_at: datetime.datetime | None = None, |
| 115 | ) -> None: |
| 116 | self.commit_id = commit_id |
| 117 | self.metadata = metadata |
| 118 | self.author = author |
| 119 | self.agent_id = agent_id |
| 120 | self.model_id = "" |
| 121 | self.toolchain_id = "" |
| 122 | self.prompt_hash = "" |
| 123 | self.committed_at = committed_at or datetime.datetime( |
| 124 | 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc |
| 125 | ) |
| 126 | |
| 127 | |
| 128 | def _signed_quorum_metadata( |
| 129 | org_handle: str, |
| 130 | threshold: int, |
| 131 | members: list[tuple[str, Ed25519PrivateKey, str]], |
| 132 | payload: str, |
| 133 | ) -> str: |
| 134 | """Build a JSON-encoded quorum metadata blob with real Ed25519 sigs.""" |
| 135 | signers: list[dict[str, str]] = [] |
| 136 | for handle, priv, pub_b64 in members: |
| 137 | sig = sign_commit_ed25519(payload, priv) |
| 138 | signers.append({"handle": handle, "public_key": pub_b64, "signature": sig}) |
| 139 | return json.dumps(build_quorum_metadata(org_handle, threshold, signers)) |
| 140 | |
| 141 | |
| 142 | @pytest.fixture |
| 143 | def identity_dir(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: |
| 144 | """Per-test identity directory wired through MUSE_IDENTITY_DIR.""" |
| 145 | d = tmp_path / "identity" |
| 146 | d.mkdir() |
| 147 | monkeypatch.setenv("MUSE_IDENTITY_DIR", str(d)) |
| 148 | yield d |
| 149 | unregister_identity_provider(PROVIDER_DOMAIN) |
| 150 | |
| 151 | |
| 152 | @pytest.fixture |
| 153 | def fs_provider(identity_dir: pathlib.Path) -> KnowtationFsIdentityProvider: |
| 154 | """Provider wired to the per-test identity_dir + registry.""" |
| 155 | return register_knowtation_identity_provider(repo_root=None) |
| 156 | |
| 157 | |
| 158 | # =========================================================================== |
| 159 | # UNIT TIER |
| 160 | # =========================================================================== |
| 161 | |
| 162 | |
| 163 | class TestHandleValidation: |
| 164 | """``is_valid_handle`` whitelists the safe character set.""" |
| 165 | |
| 166 | @pytest.mark.parametrize("handle", [ |
| 167 | "alice", "@my-org", "a.b.c", "user_42", "@root_org-1", |
| 168 | ]) |
| 169 | def test_valid(self, handle: str) -> None: |
| 170 | assert is_valid_handle(handle) is True |
| 171 | |
| 172 | @pytest.mark.parametrize("handle", [ |
| 173 | "", " ", "alice/bob", "@..", "@/etc/passwd", "alice\x00", |
| 174 | "very" + "long" * 50, None, 42, ["alice"], |
| 175 | ]) |
| 176 | def test_invalid(self, handle: object) -> None: |
| 177 | assert is_valid_handle(handle) is False |
| 178 | |
| 179 | |
| 180 | class TestBuildQuorumMetadata: |
| 181 | """build_quorum_metadata enforces shape contracts.""" |
| 182 | |
| 183 | def test_minimal_valid(self) -> None: |
| 184 | meta = build_quorum_metadata( |
| 185 | "@org", 1, |
| 186 | [{"handle": "alice", "public_key": "pk", "signature": "sig"}], |
| 187 | ) |
| 188 | assert meta["org_handle"] == "@org" |
| 189 | assert meta["threshold"] == 1 |
| 190 | assert len(meta["signers"]) == 1 |
| 191 | |
| 192 | def test_rejects_non_org_handle(self) -> None: |
| 193 | with pytest.raises(ValueError): |
| 194 | build_quorum_metadata( |
| 195 | "alice", 1, |
| 196 | [{"handle": "alice", "public_key": "pk", "signature": "sig"}], |
| 197 | ) |
| 198 | |
| 199 | def test_rejects_zero_threshold(self) -> None: |
| 200 | with pytest.raises(ValueError): |
| 201 | build_quorum_metadata( |
| 202 | "@org", 0, |
| 203 | [{"handle": "alice", "public_key": "pk", "signature": "sig"}], |
| 204 | ) |
| 205 | |
| 206 | def test_rejects_duplicate_signer(self) -> None: |
| 207 | with pytest.raises(ValueError, match="more than once"): |
| 208 | build_quorum_metadata( |
| 209 | "@org", 1, [ |
| 210 | {"handle": "alice", "public_key": "pk", "signature": "s"}, |
| 211 | {"handle": "alice", "public_key": "pk", "signature": "s"}, |
| 212 | ], |
| 213 | ) |
| 214 | |
| 215 | def test_rejects_empty_signature(self) -> None: |
| 216 | with pytest.raises(ValueError, match="signature"): |
| 217 | build_quorum_metadata( |
| 218 | "@org", 1, |
| 219 | [{"handle": "alice", "public_key": "pk", "signature": ""}], |
| 220 | ) |
| 221 | |
| 222 | |
| 223 | class TestIdentityRecordCoercion: |
| 224 | """KnowtationFsIdentityProvider validates record shape strictly.""" |
| 225 | |
| 226 | def test_round_trip(self, identity_dir: pathlib.Path, |
| 227 | fs_provider: KnowtationFsIdentityProvider) -> None: |
| 228 | _write_identity(identity_dir, "alice", kind="human", public_key="pk-alice") |
| 229 | rec = fs_provider.get_identity("alice") |
| 230 | assert rec is not None |
| 231 | assert rec["handle"] == "alice" |
| 232 | assert rec["kind"] == "human" |
| 233 | assert rec["public_key"] == "pk-alice" |
| 234 | |
| 235 | def test_unknown_returns_none(self, identity_dir: pathlib.Path, |
| 236 | fs_provider: KnowtationFsIdentityProvider) -> None: |
| 237 | assert fs_provider.get_identity("ghost") is None |
| 238 | |
| 239 | def test_handle_mismatch_raises(self, identity_dir: pathlib.Path, |
| 240 | fs_provider: KnowtationFsIdentityProvider) -> None: |
| 241 | bad = json.dumps({"handle": "evil", "kind": "human", "public_key": "pk"}) |
| 242 | (identity_dir / "alice.json").write_text(bad, encoding="utf-8") |
| 243 | with pytest.raises(AttestationVerifyError) as excinfo: |
| 244 | fs_provider.get_identity("alice") |
| 245 | assert excinfo.value.code == "HANDLE_MISMATCH" |
| 246 | |
| 247 | def test_invalid_kind_raises(self, identity_dir: pathlib.Path, |
| 248 | fs_provider: KnowtationFsIdentityProvider) -> None: |
| 249 | bad = json.dumps({"handle": "alice", "kind": "robot", "public_key": "pk"}) |
| 250 | (identity_dir / "alice.json").write_text(bad, encoding="utf-8") |
| 251 | with pytest.raises(AttestationVerifyError) as excinfo: |
| 252 | fs_provider.get_identity("alice") |
| 253 | assert excinfo.value.code == "INVALID_KIND" |
| 254 | |
| 255 | def test_threshold_exceeds_members_raises( |
| 256 | self, identity_dir: pathlib.Path, |
| 257 | fs_provider: KnowtationFsIdentityProvider, |
| 258 | ) -> None: |
| 259 | _write_identity( |
| 260 | identity_dir, "@org", kind="org", |
| 261 | members=["alice", "bob"], quorum_threshold=5, |
| 262 | ) |
| 263 | with pytest.raises(AttestationVerifyError) as excinfo: |
| 264 | fs_provider.get_identity("@org") |
| 265 | assert excinfo.value.code == "INVALID_THRESHOLD" |
| 266 | |
| 267 | |
| 268 | # =========================================================================== |
| 269 | # UNIT TIER — identity-chain walker |
| 270 | # =========================================================================== |
| 271 | |
| 272 | |
| 273 | class TestIdentityChain: |
| 274 | """verify_identity_chain handles cycles, depth, and revocation.""" |
| 275 | |
| 276 | def test_simple_org_walks_members( |
| 277 | self, identity_dir: pathlib.Path, |
| 278 | fs_provider: KnowtationFsIdentityProvider, |
| 279 | ) -> None: |
| 280 | _write_identity(identity_dir, "alice", kind="human", public_key="pk-a") |
| 281 | _write_identity(identity_dir, "bob", kind="human", public_key="pk-b") |
| 282 | _write_identity( |
| 283 | identity_dir, "@org", kind="org", |
| 284 | members=["alice", "bob"], quorum_threshold=2, |
| 285 | ) |
| 286 | result = verify_identity_chain("@org", fs_provider) |
| 287 | assert result["valid"] is True |
| 288 | assert result["depth"] == 2 |
| 289 | assert result["cycle_detected"] is False |
| 290 | |
| 291 | def test_cycle_detected( |
| 292 | self, identity_dir: pathlib.Path, |
| 293 | fs_provider: KnowtationFsIdentityProvider, |
| 294 | ) -> None: |
| 295 | _write_identity( |
| 296 | identity_dir, "@org-a", kind="org", |
| 297 | members=["@org-b"], quorum_threshold=1, |
| 298 | ) |
| 299 | _write_identity( |
| 300 | identity_dir, "@org-b", kind="org", |
| 301 | members=["@org-a"], quorum_threshold=1, |
| 302 | ) |
| 303 | result = verify_identity_chain("@org-a", fs_provider) |
| 304 | assert result["valid"] is False |
| 305 | assert result["cycle_detected"] is True |
| 306 | assert "CYCLE_DETECTED" in result["reason"] |
| 307 | |
| 308 | def test_depth_limit_enforced( |
| 309 | self, identity_dir: pathlib.Path, |
| 310 | fs_provider: KnowtationFsIdentityProvider, |
| 311 | ) -> None: |
| 312 | # Build a chain @o0 → @o1 → @o2 → ... → @o5; depth_limit=3 trips. |
| 313 | for i in range(5): |
| 314 | _write_identity( |
| 315 | identity_dir, f"@o{i}", kind="org", |
| 316 | members=[f"@o{i+1}"], quorum_threshold=1, |
| 317 | ) |
| 318 | _write_identity(identity_dir, "@o5", kind="human", public_key="pk") |
| 319 | result = verify_identity_chain("@o0", fs_provider, depth_limit=3) |
| 320 | assert result["valid"] is False |
| 321 | assert "DEPTH_EXCEEDED" in result["reason"] |
| 322 | |
| 323 | def test_revoked_member_invalidates( |
| 324 | self, identity_dir: pathlib.Path, |
| 325 | fs_provider: KnowtationFsIdentityProvider, |
| 326 | ) -> None: |
| 327 | _write_identity( |
| 328 | identity_dir, "alice", kind="human", public_key="pk-a", |
| 329 | revoked_at="2026-01-01T00:00:00Z", |
| 330 | ) |
| 331 | _write_identity( |
| 332 | identity_dir, "@org", kind="org", |
| 333 | members=["alice"], quorum_threshold=1, |
| 334 | ) |
| 335 | result = verify_identity_chain("@org", fs_provider) |
| 336 | assert result["valid"] is False |
| 337 | assert "REVOKED" in result["reason"] |
| 338 | |
| 339 | |
| 340 | # =========================================================================== |
| 341 | # INTEGRATION TIER — verify_quorum_signatures end-to-end |
| 342 | # =========================================================================== |
| 343 | |
| 344 | |
| 345 | class TestQuorumVerificationHappyPath: |
| 346 | """3-member 2-of-3 org with all permutations of 2 valid signers.""" |
| 347 | |
| 348 | @pytest.fixture |
| 349 | def setup(self, identity_dir: pathlib.Path, |
| 350 | fs_provider: KnowtationFsIdentityProvider) -> dict[str, Any]: |
| 351 | members: dict[str, tuple[Ed25519PrivateKey, str]] = {} |
| 352 | for h in ("alice", "bob", "carol"): |
| 353 | priv, pub = _new_keypair() |
| 354 | members[h] = (priv, pub) |
| 355 | _write_identity(identity_dir, h, kind="human", public_key=pub) |
| 356 | _write_identity( |
| 357 | identity_dir, "@team", kind="org", |
| 358 | members=list(members), quorum_threshold=2, |
| 359 | ) |
| 360 | commit_id = "a" * 64 |
| 361 | committed_at = datetime.datetime( |
| 362 | 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc |
| 363 | ) |
| 364 | payload = provenance_payload( |
| 365 | commit_id, author="aaron", agent_id="agent-x", |
| 366 | committed_at=committed_at.isoformat(), |
| 367 | ) |
| 368 | return { |
| 369 | "members": members, "commit_id": commit_id, |
| 370 | "committed_at": committed_at, "payload": payload, |
| 371 | } |
| 372 | |
| 373 | @pytest.mark.parametrize("signer_pair", [ |
| 374 | ("alice", "bob"), |
| 375 | ("alice", "carol"), |
| 376 | ("bob", "carol"), |
| 377 | ]) |
| 378 | def test_each_pair_meets_threshold( |
| 379 | self, setup: dict[str, Any], signer_pair: tuple[str, str], |
| 380 | ) -> None: |
| 381 | members = [ |
| 382 | (h, setup["members"][h][0], setup["members"][h][1]) |
| 383 | for h in signer_pair |
| 384 | ] |
| 385 | meta_blob = _signed_quorum_metadata( |
| 386 | "@team", 2, members, setup["payload"], |
| 387 | ) |
| 388 | commit = _FakeCommit( |
| 389 | setup["commit_id"], {"quorum": meta_blob}, |
| 390 | author="aaron", agent_id="agent-x", |
| 391 | committed_at=setup["committed_at"], |
| 392 | ) |
| 393 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 394 | assert result["valid"] is True, result.get("reason") |
| 395 | assert result["valid_signers"] == 2 |
| 396 | assert result["threshold"] == 2 |
| 397 | assert result["org_handle"] == "@team" |
| 398 | |
| 399 | |
| 400 | # =========================================================================== |
| 401 | # SECURITY TIER (mandatory per Rule #0) |
| 402 | # =========================================================================== |
| 403 | |
| 404 | |
| 405 | class TestQuorumSecurity: |
| 406 | """All threats from the Phase 7.7 spec — never silently degrade.""" |
| 407 | |
| 408 | @pytest.fixture |
| 409 | def env(self, identity_dir: pathlib.Path, |
| 410 | fs_provider: KnowtationFsIdentityProvider) -> dict[str, Any]: |
| 411 | members: dict[str, tuple[Ed25519PrivateKey, str]] = {} |
| 412 | for h in ("alice", "bob", "carol"): |
| 413 | priv, pub = _new_keypair() |
| 414 | members[h] = (priv, pub) |
| 415 | _write_identity(identity_dir, h, kind="human", public_key=pub) |
| 416 | _write_identity( |
| 417 | identity_dir, "@team", kind="org", |
| 418 | members=list(members), quorum_threshold=2, |
| 419 | ) |
| 420 | commit_id = "b" * 64 |
| 421 | committed_at = datetime.datetime( |
| 422 | 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc |
| 423 | ) |
| 424 | payload = provenance_payload( |
| 425 | commit_id, author="aaron", agent_id="agent-x", |
| 426 | committed_at=committed_at.isoformat(), |
| 427 | ) |
| 428 | return { |
| 429 | "identity_dir": identity_dir, |
| 430 | "members": members, "commit_id": commit_id, |
| 431 | "committed_at": committed_at, "payload": payload, |
| 432 | } |
| 433 | |
| 434 | def test_partial_quorum_forgery_rejected(self, env: dict[str, Any]) -> None: |
| 435 | """Only floor(threshold-1)=1 valid sig → MUST be rejected.""" |
| 436 | signer = ("alice", env["members"]["alice"][0], env["members"]["alice"][1]) |
| 437 | meta_blob = _signed_quorum_metadata("@team", 2, [signer], env["payload"]) |
| 438 | commit = _FakeCommit( |
| 439 | env["commit_id"], {"quorum": meta_blob}, |
| 440 | author="aaron", agent_id="agent-x", |
| 441 | committed_at=env["committed_at"], |
| 442 | ) |
| 443 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 444 | assert result["valid"] is False |
| 445 | assert "BELOW_THRESHOLD" in result["reason"] |
| 446 | |
| 447 | def test_unknown_signer_rejected(self, env: dict[str, Any]) -> None: |
| 448 | """A signer outside the org's members list → reject (collusion attempt).""" |
| 449 | outsider_priv, outsider_pub = _new_keypair() |
| 450 | # Write an identity for the outsider so the per-member fetch succeeds — |
| 451 | # the rejection MUST come from the membership check, not a missing record. |
| 452 | _write_identity( |
| 453 | env["identity_dir"], "mallory", kind="human", public_key=outsider_pub, |
| 454 | ) |
| 455 | members = [ |
| 456 | ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), |
| 457 | ("mallory", outsider_priv, outsider_pub), |
| 458 | ] |
| 459 | meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) |
| 460 | commit = _FakeCommit( |
| 461 | env["commit_id"], {"quorum": meta_blob}, |
| 462 | author="aaron", agent_id="agent-x", |
| 463 | committed_at=env["committed_at"], |
| 464 | ) |
| 465 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 466 | assert result["valid"] is False |
| 467 | assert "UNKNOWN_SIGNER" in result["reason"] |
| 468 | |
| 469 | def test_signature_for_different_commit_rejected(self, env: dict[str, Any]) -> None: |
| 470 | """Replay defence: sig over a different commit_id MUST NOT verify.""" |
| 471 | wrong_payload = provenance_payload( |
| 472 | "f" * 64, # different commit_id |
| 473 | author="aaron", agent_id="agent-x", |
| 474 | committed_at=env["committed_at"].isoformat(), |
| 475 | ) |
| 476 | members = [ |
| 477 | ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), |
| 478 | ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), |
| 479 | ] |
| 480 | meta_blob = _signed_quorum_metadata("@team", 2, members, wrong_payload) |
| 481 | commit = _FakeCommit( |
| 482 | env["commit_id"], {"quorum": meta_blob}, |
| 483 | author="aaron", agent_id="agent-x", |
| 484 | committed_at=env["committed_at"], |
| 485 | ) |
| 486 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 487 | assert result["valid"] is False |
| 488 | assert "SIGNATURE_INVALID" in result["reason"] |
| 489 | |
| 490 | def test_public_key_mismatch_rejected(self, env: dict[str, Any]) -> None: |
| 491 | """Forged public_key in metadata MUST NOT bypass canonical key check.""" |
| 492 | fake_priv, fake_pub = _new_keypair() |
| 493 | sig = sign_commit_ed25519(env["payload"], fake_priv) |
| 494 | # The handle is real, the signature is valid for the *fake* key, but |
| 495 | # the canonical record's public_key is the real alice's. Reject. |
| 496 | meta_blob = json.dumps(build_quorum_metadata( |
| 497 | "@team", 2, [ |
| 498 | {"handle": "alice", "public_key": fake_pub, "signature": sig}, |
| 499 | {"handle": "bob", |
| 500 | "public_key": env["members"]["bob"][1], |
| 501 | "signature": sign_commit_ed25519(env["payload"], env["members"]["bob"][0])}, |
| 502 | ], |
| 503 | )) |
| 504 | commit = _FakeCommit( |
| 505 | env["commit_id"], {"quorum": meta_blob}, |
| 506 | author="aaron", agent_id="agent-x", |
| 507 | committed_at=env["committed_at"], |
| 508 | ) |
| 509 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 510 | assert result["valid"] is False |
| 511 | assert "SIGNATURE_INVALID" in result["reason"] |
| 512 | assert "alice" in result["reason"] |
| 513 | |
| 514 | def test_revoked_member_does_not_count( |
| 515 | self, identity_dir: pathlib.Path, env: dict[str, Any], |
| 516 | ) -> None: |
| 517 | """Revoked member's sig must NOT count toward quorum (and must reject).""" |
| 518 | # Revoke alice but keep her real public_key for reproducibility. |
| 519 | _write_identity( |
| 520 | identity_dir, "alice", kind="human", |
| 521 | public_key=env["members"]["alice"][1], |
| 522 | revoked_at="2026-04-01T00:00:00Z", |
| 523 | ) |
| 524 | members = [ |
| 525 | ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), |
| 526 | ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), |
| 527 | ] |
| 528 | meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) |
| 529 | commit = _FakeCommit( |
| 530 | env["commit_id"], {"quorum": meta_blob}, |
| 531 | author="aaron", agent_id="agent-x", |
| 532 | committed_at=env["committed_at"], |
| 533 | ) |
| 534 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 535 | assert result["valid"] is False |
| 536 | # The chain walker rejects revoked members during the org walk so the |
| 537 | # surfaced reason starts with "REVOKED" rather than the per-signer |
| 538 | # "REVOKED_MEMBER" code — both code paths reject revoked members. |
| 539 | assert "REVOKED" in result["reason"] |
| 540 | |
| 541 | def test_cycle_introducing_membership_rejected( |
| 542 | self, identity_dir: pathlib.Path, |
| 543 | ) -> None: |
| 544 | """Defensive cycle detection at verify time.""" |
| 545 | register_knowtation_identity_provider(repo_root=None) |
| 546 | _write_identity( |
| 547 | identity_dir, "@org-a", kind="org", |
| 548 | members=["@org-b"], quorum_threshold=1, |
| 549 | ) |
| 550 | _write_identity( |
| 551 | identity_dir, "@org-b", kind="org", |
| 552 | members=["@org-a"], quorum_threshold=1, |
| 553 | ) |
| 554 | # No need to even sign — the chain walker rejects before sig checks. |
| 555 | commit = _FakeCommit( |
| 556 | "c" * 64, |
| 557 | {"quorum": json.dumps({ |
| 558 | "org_handle": "@org-a", "threshold": 1, |
| 559 | "signers": [{"handle": "alice", "public_key": "pk", "signature": "s"}], |
| 560 | })}, |
| 561 | ) |
| 562 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 563 | assert result["valid"] is False |
| 564 | assert "CYCLE_DETECTED" in result["reason"] |
| 565 | |
| 566 | def test_provider_missing_is_partial(self, env: dict[str, Any]) -> None: |
| 567 | """No identity provider registered → partial=True (NEVER silent True).""" |
| 568 | unregister_identity_provider(PROVIDER_DOMAIN) |
| 569 | members = [ |
| 570 | ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), |
| 571 | ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), |
| 572 | ] |
| 573 | meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) |
| 574 | commit = _FakeCommit( |
| 575 | env["commit_id"], {"quorum": meta_blob}, |
| 576 | author="aaron", agent_id="agent-x", |
| 577 | committed_at=env["committed_at"], |
| 578 | ) |
| 579 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 580 | assert result["valid"] is False |
| 581 | assert result["partial"] is True |
| 582 | assert "IDENTITY_PROVIDER_MISSING" in result["reason"] |
| 583 | |
| 584 | def test_no_quorum_metadata_rejected(self) -> None: |
| 585 | """Commit without metadata['quorum'] → reject without provider lookup.""" |
| 586 | commit = _FakeCommit("d" * 64, {}) |
| 587 | result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) |
| 588 | assert result["valid"] is False |
| 589 | assert "NO_QUORUM_METADATA" in result["reason"] |
| 590 | |
| 591 | def test_path_traversal_handle_rejected( |
| 592 | self, identity_dir: pathlib.Path, |
| 593 | fs_provider: KnowtationFsIdentityProvider, |
| 594 | ) -> None: |
| 595 | """Handles with ``..`` / ``/`` MUST raise — never read arbitrary files.""" |
| 596 | with pytest.raises(AttestationVerifyError): |
| 597 | fs_provider.get_identity("../etc/passwd") |
| 598 | with pytest.raises(AttestationVerifyError): |
| 599 | fs_provider.get_identity("..\\..\\windows\\system32") |
| 600 | |
| 601 | def test_oversized_record_rejected( |
| 602 | self, identity_dir: pathlib.Path, |
| 603 | fs_provider: KnowtationFsIdentityProvider, |
| 604 | ) -> None: |
| 605 | """Identity files > _MAX_RECORD_BYTES MUST raise.""" |
| 606 | big = json.dumps({ |
| 607 | "handle": "alice", "kind": "human", "public_key": "x", |
| 608 | "padding": "A" * 20_000, |
| 609 | }) |
| 610 | (identity_dir / "alice.json").write_text(big, encoding="utf-8") |
| 611 | with pytest.raises(AttestationVerifyError) as excinfo: |
| 612 | fs_provider.get_identity("alice") |
| 613 | assert excinfo.value.code == "RECORD_TOO_LARGE" |
| 614 | |
| 615 | |
| 616 | # =========================================================================== |
| 617 | # UNIT TIER — registry + concurrency |
| 618 | # =========================================================================== |
| 619 | |
| 620 | |
| 621 | class TestRegistryConcurrency: |
| 622 | """Identity provider registry is thread-safe under concurrent register/get.""" |
| 623 | |
| 624 | def test_register_and_get(self, identity_dir: pathlib.Path) -> None: |
| 625 | provider = register_knowtation_identity_provider(repo_root=None) |
| 626 | from muse.core.attestation import get_identity_provider |
| 627 | retrieved = get_identity_provider(PROVIDER_DOMAIN) |
| 628 | assert retrieved is provider |
| 629 | |
| 630 | def test_unregister(self, identity_dir: pathlib.Path) -> None: |
| 631 | register_knowtation_identity_provider(repo_root=None) |
| 632 | assert unregister_identity_provider(PROVIDER_DOMAIN) is True |
| 633 | assert unregister_identity_provider(PROVIDER_DOMAIN) is False |
| 634 | |
| 635 | def test_concurrent_register_get(self, identity_dir: pathlib.Path) -> None: |
| 636 | from muse.core.attestation import get_identity_provider |
| 637 | |
| 638 | errors: list[Exception] = [] |
| 639 | |
| 640 | def worker() -> None: |
| 641 | try: |
| 642 | for _ in range(100): |
| 643 | register_knowtation_identity_provider(repo_root=None) |
| 644 | get_identity_provider(PROVIDER_DOMAIN) |
| 645 | unregister_identity_provider(PROVIDER_DOMAIN) |
| 646 | except Exception as e: |
| 647 | errors.append(e) |
| 648 | |
| 649 | threads = [threading.Thread(target=worker) for _ in range(8)] |
| 650 | for t in threads: |
| 651 | t.start() |
| 652 | for t in threads: |
| 653 | t.join() |
| 654 | assert errors == [] |
| 655 | |
| 656 | |
| 657 | # =========================================================================== |
| 658 | # END-TO-END TIER — muse commit --quorum-signers / verify-commit --verify-identity |
| 659 | # =========================================================================== |
| 660 | |
| 661 | |
| 662 | def _write_pem(path: pathlib.Path, priv: Ed25519PrivateKey) -> None: |
| 663 | """Write a PEM-encoded private key to *path*.""" |
| 664 | pem = priv.private_bytes( |
| 665 | encoding=Encoding.PEM, |
| 666 | format=PrivateFormat.PKCS8, |
| 667 | encryption_algorithm=NoEncryption(), |
| 668 | ) |
| 669 | path.write_bytes(pem) |
| 670 | |
| 671 | |
| 672 | def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 673 | """Initialise a knowtation-domain repo at *tmp_path* and chdir into it.""" |
| 674 | import argparse as _argparse |
| 675 | import os as _os |
| 676 | _os.chdir(tmp_path) |
| 677 | from muse.cli.commands.init import run as init_run |
| 678 | args = _argparse.Namespace( |
| 679 | bare=False, template=None, default_branch="main", |
| 680 | force=False, domain="knowtation", as_json=False, |
| 681 | ) |
| 682 | init_run(args) |
| 683 | (tmp_path / "notes").mkdir(exist_ok=True) |
| 684 | (tmp_path / "notes" / "seed.md").write_text("# seed\n", encoding="utf-8") |
| 685 | return tmp_path |
| 686 | |
| 687 | |
| 688 | def _run_commit(tmp_path: pathlib.Path, **overrides: Any) -> dict[str, Any]: |
| 689 | """Run muse commit (json mode) with sensible defaults + overrides. |
| 690 | |
| 691 | Captures stdout AND stderr so failed commits surface their reason via |
| 692 | ``AssertionError`` (otherwise the test sees only ``ExitCode.USER_ERROR`` |
| 693 | with no diagnostic). |
| 694 | """ |
| 695 | import argparse as _argparse |
| 696 | import io as _io |
| 697 | import sys as _sys |
| 698 | from muse.cli.commands import commit as commit_cmd |
| 699 | |
| 700 | base = dict( |
| 701 | message="first", allow_empty=False, dry_run=False, |
| 702 | section=None, track=None, emotion=None, author=None, |
| 703 | agent_id=None, model_id=None, toolchain_id=None, |
| 704 | event_type=None, meta=None, |
| 705 | sign=False, attest=False, attest_config=None, |
| 706 | quorum_signers=None, quorum_org=None, |
| 707 | quorum_threshold=None, quorum_keys=None, |
| 708 | fmt="json", |
| 709 | ) |
| 710 | base.update(overrides) |
| 711 | out_buf = _io.StringIO() |
| 712 | err_buf = _io.StringIO() |
| 713 | real_out, real_err = _sys.stdout, _sys.stderr |
| 714 | raised: BaseException | None = None |
| 715 | try: |
| 716 | _sys.stdout, _sys.stderr = out_buf, err_buf |
| 717 | try: |
| 718 | commit_cmd.run(_argparse.Namespace(**base)) |
| 719 | except BaseException as exc: |
| 720 | raised = exc |
| 721 | finally: |
| 722 | _sys.stdout, _sys.stderr = real_out, real_err |
| 723 | out_text = out_buf.getvalue() |
| 724 | err_text = err_buf.getvalue() |
| 725 | if raised is not None: |
| 726 | # Attach captured buffers to the exception so the test sees them. |
| 727 | raised.captured_stdout = out_text # type: ignore[attr-defined] |
| 728 | raised.captured_stderr = err_text # type: ignore[attr-defined] |
| 729 | raise raised |
| 730 | payload_lines = [ |
| 731 | line for line in out_text.splitlines() if line.strip().startswith("{") |
| 732 | ] |
| 733 | assert payload_lines, ( |
| 734 | f"commit produced no JSON output. stdout={out_text!r} stderr={err_text!r}" |
| 735 | ) |
| 736 | return json.loads(payload_lines[-1]) |
| 737 | |
| 738 | |
| 739 | class TestCommitQuorumE2E: |
| 740 | """End-to-end: muse commit --quorum-signers writes the quorum metadata.""" |
| 741 | |
| 742 | def test_three_member_two_of_three_quorum_round_trip( |
| 743 | self, |
| 744 | tmp_path: pathlib.Path, |
| 745 | identity_dir: pathlib.Path, |
| 746 | monkeypatch: pytest.MonkeyPatch, |
| 747 | ) -> None: |
| 748 | # Prepare 3 members with PEMs. |
| 749 | keys: dict[str, tuple[Ed25519PrivateKey, str]] = {} |
| 750 | for h in ("alice", "bob", "carol"): |
| 751 | priv, pub = _new_keypair() |
| 752 | keys[h] = (priv, pub) |
| 753 | pem_path = tmp_path / f"{h}.pem" |
| 754 | _write_pem(pem_path, priv) |
| 755 | monkeypatch.setenv( |
| 756 | f"MUSE_QUORUM_KEY_{h.upper()}_PATH", str(pem_path) |
| 757 | ) |
| 758 | _write_identity(identity_dir, h, kind="human", public_key=pub) |
| 759 | _write_identity( |
| 760 | identity_dir, "@team", kind="org", |
| 761 | members=list(keys), quorum_threshold=2, |
| 762 | ) |
| 763 | register_knowtation_identity_provider(repo_root=None) |
| 764 | |
| 765 | # Run a commit with two co-signers. commit_cmd.run() does NOT |
| 766 | # raise SystemExit on the success path (just returns), so the test |
| 767 | # also tolerates the non-exit case. |
| 768 | repo = _init_repo(tmp_path) |
| 769 | result_payload: dict[str, Any] | None = None |
| 770 | try: |
| 771 | result_payload = _run_commit( |
| 772 | repo, message="quorum-test", |
| 773 | quorum_signers="alice,bob", quorum_org="@team", |
| 774 | quorum_threshold=2, |
| 775 | ) |
| 776 | except SystemExit as exc: |
| 777 | stderr_text = getattr(exc, "captured_stderr", "") |
| 778 | stdout_text = getattr(exc, "captured_stdout", "") |
| 779 | assert exc.code in (0, None), ( |
| 780 | f"commit failed: exit={exc.code}\n" |
| 781 | f"stderr={stderr_text!r}\nstdout={stdout_text!r}" |
| 782 | ) |
| 783 | assert result_payload is None or "commit_id" in result_payload |
| 784 | |
| 785 | # Read the freshly-written commit and verify the quorum metadata |
| 786 | # round-trips through verify_quorum_signatures. |
| 787 | from muse.core.store import get_head_commit_id, read_commit |
| 788 | head = get_head_commit_id(repo, "main") |
| 789 | assert head is not None |
| 790 | commit = read_commit(repo, head) |
| 791 | assert commit is not None |
| 792 | assert "quorum" in commit.metadata |
| 793 | result = verify_quorum_signatures(commit, repo) |
| 794 | assert result["valid"] is True, result.get("reason") |
| 795 | assert result["valid_signers"] == 2 |
| 796 | assert result["threshold"] == 2 |
| 797 | |
| 798 | def test_missing_quorum_key_aborts_commit( |
| 799 | self, |
| 800 | tmp_path: pathlib.Path, |
| 801 | identity_dir: pathlib.Path, |
| 802 | monkeypatch: pytest.MonkeyPatch, |
| 803 | ) -> None: |
| 804 | """Signature stripping defence: missing key → ExitCode.USER_ERROR.""" |
| 805 | # Only one member has a key on disk; alice is intentionally missing. |
| 806 | bob_priv, bob_pub = _new_keypair() |
| 807 | bob_path = tmp_path / "bob.pem" |
| 808 | _write_pem(bob_path, bob_priv) |
| 809 | monkeypatch.setenv("MUSE_QUORUM_KEY_BOB_PATH", str(bob_path)) |
| 810 | _write_identity(identity_dir, "alice", kind="human", public_key="pk-a") |
| 811 | _write_identity(identity_dir, "bob", kind="human", public_key=bob_pub) |
| 812 | _write_identity( |
| 813 | identity_dir, "@team", kind="org", |
| 814 | members=["alice", "bob"], quorum_threshold=2, |
| 815 | ) |
| 816 | |
| 817 | # Make sure alice's default ~/.muse/keys path is NOT inadvertently |
| 818 | # found — point HOME at the tmp dir to isolate. |
| 819 | monkeypatch.setenv("HOME", str(tmp_path)) |
| 820 | |
| 821 | repo = _init_repo(tmp_path) |
| 822 | with pytest.raises(SystemExit) as excinfo: |
| 823 | _run_commit( |
| 824 | repo, message="missing-key", |
| 825 | quorum_signers="alice,bob", quorum_org="@team", |
| 826 | quorum_threshold=2, |
| 827 | ) |
| 828 | # ExitCode.USER_ERROR == 1 |
| 829 | assert excinfo.value.code == 1 |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
26 days ago