"""Tests for Phase 7.7 — org-quorum signing via the identity domain. Coverage tiers (Rule #0) ------------------------ * unit — handle validation, identity-record shape coercion, filename encoding, build_quorum_metadata edge cases, quorum-math threshold edges, identity-chain cycle detection. * integration — KnowtationFsIdentityProvider round-trip (write JSON → read back), 3-member 2-of-3 quorum with all permutations of two valid signers verifying. * security — partial-quorum forgery rejected; signer collusion below threshold rejected; signature stripping detected at commit time; signer public_key mismatch rejected; cycle-introducing membership detected; depth-bounded recursion; revoked members do not count; path-traversal handles rejected; provider missing → partial=True (never silent True). * end-to-end — N/A (covered by the CLI integration in test_cli_workflow once available; this suite stays focused on the engine + provider). """ from __future__ import annotations import base64 import datetime import json import pathlib import threading from typing import Any import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.serialization import ( Encoding, NoEncryption, PrivateFormat, PublicFormat, ) from muse.core.attestation import ( AttestationVerifyError, IdentityRecord, register_identity_provider, unregister_identity_provider, ) from muse.core.provenance import ( encode_public_key, provenance_payload, sign_commit_ed25519, ) from muse.core.verify import ( DEFAULT_IDENTITY_DEPTH_LIMIT, QuorumVerifyResult, verify_identity_chain, verify_quorum_signatures, ) from muse.plugins.knowtation.quorum import ( KnowtationFsIdentityProvider, PROVIDER_DOMAIN, build_quorum_metadata, is_valid_handle, register_knowtation_identity_provider, ) # --------------------------------------------------------------------------- # Test helpers # --------------------------------------------------------------------------- def _new_keypair() -> tuple[Ed25519PrivateKey, str]: """Return ``(private_key, base64url_public_key_no_padding)``.""" priv = Ed25519PrivateKey.generate() _raw, pub_b64 = encode_public_key(priv) return priv, pub_b64 def _write_identity( dir_: pathlib.Path, handle: str, *, kind: str, public_key: str = "", members: list[str] | None = None, quorum_threshold: int = 0, revoked_at: str = "", ) -> None: """Persist an identity record JSON file under *dir_*.""" dir_.mkdir(parents=True, exist_ok=True) fname = ("_at_" + handle[1:]) if handle.startswith("@") else handle body: dict[str, Any] = { "handle": handle, "kind": kind, "public_key": public_key, } if members is not None: body["members"] = members if quorum_threshold: body["quorum_threshold"] = quorum_threshold if revoked_at: body["revoked_at"] = revoked_at (dir_ / f"{fname}.json").write_text(json.dumps(body), encoding="utf-8") class _FakeCommit: """Minimal CommitRecord-shaped duck for verify_quorum_signatures.""" def __init__( self, commit_id: str, metadata: dict[str, Any], *, author: str = "", agent_id: str = "", committed_at: datetime.datetime | None = None, ) -> None: self.commit_id = commit_id self.metadata = metadata self.author = author self.agent_id = agent_id self.model_id = "" self.toolchain_id = "" self.prompt_hash = "" self.committed_at = committed_at or datetime.datetime( 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc ) def _signed_quorum_metadata( org_handle: str, threshold: int, members: list[tuple[str, Ed25519PrivateKey, str]], payload: str, ) -> str: """Build a JSON-encoded quorum metadata blob with real Ed25519 sigs.""" signers: list[dict[str, str]] = [] for handle, priv, pub_b64 in members: sig = sign_commit_ed25519(payload, priv) signers.append({"handle": handle, "public_key": pub_b64, "signature": sig}) return json.dumps(build_quorum_metadata(org_handle, threshold, signers)) @pytest.fixture def identity_dir(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Per-test identity directory wired through MUSE_IDENTITY_DIR.""" d = tmp_path / "identity" d.mkdir() monkeypatch.setenv("MUSE_IDENTITY_DIR", str(d)) yield d unregister_identity_provider(PROVIDER_DOMAIN) @pytest.fixture def fs_provider(identity_dir: pathlib.Path) -> KnowtationFsIdentityProvider: """Provider wired to the per-test identity_dir + registry.""" return register_knowtation_identity_provider(repo_root=None) # =========================================================================== # UNIT TIER # =========================================================================== class TestHandleValidation: """``is_valid_handle`` whitelists the safe character set.""" @pytest.mark.parametrize("handle", [ "alice", "@my-org", "a.b.c", "user_42", "@root_org-1", ]) def test_valid(self, handle: str) -> None: assert is_valid_handle(handle) is True @pytest.mark.parametrize("handle", [ "", " ", "alice/bob", "@..", "@/etc/passwd", "alice\x00", "very" + "long" * 50, None, 42, ["alice"], ]) def test_invalid(self, handle: object) -> None: assert is_valid_handle(handle) is False class TestBuildQuorumMetadata: """build_quorum_metadata enforces shape contracts.""" def test_minimal_valid(self) -> None: meta = build_quorum_metadata( "@org", 1, [{"handle": "alice", "public_key": "pk", "signature": "sig"}], ) assert meta["org_handle"] == "@org" assert meta["threshold"] == 1 assert len(meta["signers"]) == 1 def test_rejects_non_org_handle(self) -> None: with pytest.raises(ValueError): build_quorum_metadata( "alice", 1, [{"handle": "alice", "public_key": "pk", "signature": "sig"}], ) def test_rejects_zero_threshold(self) -> None: with pytest.raises(ValueError): build_quorum_metadata( "@org", 0, [{"handle": "alice", "public_key": "pk", "signature": "sig"}], ) def test_rejects_duplicate_signer(self) -> None: with pytest.raises(ValueError, match="more than once"): build_quorum_metadata( "@org", 1, [ {"handle": "alice", "public_key": "pk", "signature": "s"}, {"handle": "alice", "public_key": "pk", "signature": "s"}, ], ) def test_rejects_empty_signature(self) -> None: with pytest.raises(ValueError, match="signature"): build_quorum_metadata( "@org", 1, [{"handle": "alice", "public_key": "pk", "signature": ""}], ) class TestIdentityRecordCoercion: """KnowtationFsIdentityProvider validates record shape strictly.""" def test_round_trip(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> None: _write_identity(identity_dir, "alice", kind="human", public_key="pk-alice") rec = fs_provider.get_identity("alice") assert rec is not None assert rec["handle"] == "alice" assert rec["kind"] == "human" assert rec["public_key"] == "pk-alice" def test_unknown_returns_none(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> None: assert fs_provider.get_identity("ghost") is None def test_handle_mismatch_raises(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> None: bad = json.dumps({"handle": "evil", "kind": "human", "public_key": "pk"}) (identity_dir / "alice.json").write_text(bad, encoding="utf-8") with pytest.raises(AttestationVerifyError) as excinfo: fs_provider.get_identity("alice") assert excinfo.value.code == "HANDLE_MISMATCH" def test_invalid_kind_raises(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> None: bad = json.dumps({"handle": "alice", "kind": "robot", "public_key": "pk"}) (identity_dir / "alice.json").write_text(bad, encoding="utf-8") with pytest.raises(AttestationVerifyError) as excinfo: fs_provider.get_identity("alice") assert excinfo.value.code == "INVALID_KIND" def test_threshold_exceeds_members_raises( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: _write_identity( identity_dir, "@org", kind="org", members=["alice", "bob"], quorum_threshold=5, ) with pytest.raises(AttestationVerifyError) as excinfo: fs_provider.get_identity("@org") assert excinfo.value.code == "INVALID_THRESHOLD" # =========================================================================== # UNIT TIER — identity-chain walker # =========================================================================== class TestIdentityChain: """verify_identity_chain handles cycles, depth, and revocation.""" def test_simple_org_walks_members( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: _write_identity(identity_dir, "alice", kind="human", public_key="pk-a") _write_identity(identity_dir, "bob", kind="human", public_key="pk-b") _write_identity( identity_dir, "@org", kind="org", members=["alice", "bob"], quorum_threshold=2, ) result = verify_identity_chain("@org", fs_provider) assert result["valid"] is True assert result["depth"] == 2 assert result["cycle_detected"] is False def test_cycle_detected( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: _write_identity( identity_dir, "@org-a", kind="org", members=["@org-b"], quorum_threshold=1, ) _write_identity( identity_dir, "@org-b", kind="org", members=["@org-a"], quorum_threshold=1, ) result = verify_identity_chain("@org-a", fs_provider) assert result["valid"] is False assert result["cycle_detected"] is True assert "CYCLE_DETECTED" in result["reason"] def test_depth_limit_enforced( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: # Build a chain @o0 → @o1 → @o2 → ... → @o5; depth_limit=3 trips. for i in range(5): _write_identity( identity_dir, f"@o{i}", kind="org", members=[f"@o{i+1}"], quorum_threshold=1, ) _write_identity(identity_dir, "@o5", kind="human", public_key="pk") result = verify_identity_chain("@o0", fs_provider, depth_limit=3) assert result["valid"] is False assert "DEPTH_EXCEEDED" in result["reason"] def test_revoked_member_invalidates( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: _write_identity( identity_dir, "alice", kind="human", public_key="pk-a", revoked_at="2026-01-01T00:00:00Z", ) _write_identity( identity_dir, "@org", kind="org", members=["alice"], quorum_threshold=1, ) result = verify_identity_chain("@org", fs_provider) assert result["valid"] is False assert "REVOKED" in result["reason"] # =========================================================================== # INTEGRATION TIER — verify_quorum_signatures end-to-end # =========================================================================== class TestQuorumVerificationHappyPath: """3-member 2-of-3 org with all permutations of 2 valid signers.""" @pytest.fixture def setup(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> dict[str, Any]: members: dict[str, tuple[Ed25519PrivateKey, str]] = {} for h in ("alice", "bob", "carol"): priv, pub = _new_keypair() members[h] = (priv, pub) _write_identity(identity_dir, h, kind="human", public_key=pub) _write_identity( identity_dir, "@team", kind="org", members=list(members), quorum_threshold=2, ) commit_id = "a" * 64 committed_at = datetime.datetime( 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc ) payload = provenance_payload( commit_id, author="aaron", agent_id="agent-x", committed_at=committed_at.isoformat(), ) return { "members": members, "commit_id": commit_id, "committed_at": committed_at, "payload": payload, } @pytest.mark.parametrize("signer_pair", [ ("alice", "bob"), ("alice", "carol"), ("bob", "carol"), ]) def test_each_pair_meets_threshold( self, setup: dict[str, Any], signer_pair: tuple[str, str], ) -> None: members = [ (h, setup["members"][h][0], setup["members"][h][1]) for h in signer_pair ] meta_blob = _signed_quorum_metadata( "@team", 2, members, setup["payload"], ) commit = _FakeCommit( setup["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=setup["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is True, result.get("reason") assert result["valid_signers"] == 2 assert result["threshold"] == 2 assert result["org_handle"] == "@team" # =========================================================================== # SECURITY TIER (mandatory per Rule #0) # =========================================================================== class TestQuorumSecurity: """All threats from the Phase 7.7 spec — never silently degrade.""" @pytest.fixture def env(self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider) -> dict[str, Any]: members: dict[str, tuple[Ed25519PrivateKey, str]] = {} for h in ("alice", "bob", "carol"): priv, pub = _new_keypair() members[h] = (priv, pub) _write_identity(identity_dir, h, kind="human", public_key=pub) _write_identity( identity_dir, "@team", kind="org", members=list(members), quorum_threshold=2, ) commit_id = "b" * 64 committed_at = datetime.datetime( 2026, 5, 13, 12, 0, 0, tzinfo=datetime.timezone.utc ) payload = provenance_payload( commit_id, author="aaron", agent_id="agent-x", committed_at=committed_at.isoformat(), ) return { "identity_dir": identity_dir, "members": members, "commit_id": commit_id, "committed_at": committed_at, "payload": payload, } def test_partial_quorum_forgery_rejected(self, env: dict[str, Any]) -> None: """Only floor(threshold-1)=1 valid sig → MUST be rejected.""" signer = ("alice", env["members"]["alice"][0], env["members"]["alice"][1]) meta_blob = _signed_quorum_metadata("@team", 2, [signer], env["payload"]) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "BELOW_THRESHOLD" in result["reason"] def test_unknown_signer_rejected(self, env: dict[str, Any]) -> None: """A signer outside the org's members list → reject (collusion attempt).""" outsider_priv, outsider_pub = _new_keypair() # Write an identity for the outsider so the per-member fetch succeeds — # the rejection MUST come from the membership check, not a missing record. _write_identity( env["identity_dir"], "mallory", kind="human", public_key=outsider_pub, ) members = [ ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), ("mallory", outsider_priv, outsider_pub), ] meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "UNKNOWN_SIGNER" in result["reason"] def test_signature_for_different_commit_rejected(self, env: dict[str, Any]) -> None: """Replay defence: sig over a different commit_id MUST NOT verify.""" wrong_payload = provenance_payload( "f" * 64, # different commit_id author="aaron", agent_id="agent-x", committed_at=env["committed_at"].isoformat(), ) members = [ ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), ] meta_blob = _signed_quorum_metadata("@team", 2, members, wrong_payload) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "SIGNATURE_INVALID" in result["reason"] def test_public_key_mismatch_rejected(self, env: dict[str, Any]) -> None: """Forged public_key in metadata MUST NOT bypass canonical key check.""" fake_priv, fake_pub = _new_keypair() sig = sign_commit_ed25519(env["payload"], fake_priv) # The handle is real, the signature is valid for the *fake* key, but # the canonical record's public_key is the real alice's. Reject. meta_blob = json.dumps(build_quorum_metadata( "@team", 2, [ {"handle": "alice", "public_key": fake_pub, "signature": sig}, {"handle": "bob", "public_key": env["members"]["bob"][1], "signature": sign_commit_ed25519(env["payload"], env["members"]["bob"][0])}, ], )) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "SIGNATURE_INVALID" in result["reason"] assert "alice" in result["reason"] def test_revoked_member_does_not_count( self, identity_dir: pathlib.Path, env: dict[str, Any], ) -> None: """Revoked member's sig must NOT count toward quorum (and must reject).""" # Revoke alice but keep her real public_key for reproducibility. _write_identity( identity_dir, "alice", kind="human", public_key=env["members"]["alice"][1], revoked_at="2026-04-01T00:00:00Z", ) members = [ ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), ] meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False # The chain walker rejects revoked members during the org walk so the # surfaced reason starts with "REVOKED" rather than the per-signer # "REVOKED_MEMBER" code — both code paths reject revoked members. assert "REVOKED" in result["reason"] def test_cycle_introducing_membership_rejected( self, identity_dir: pathlib.Path, ) -> None: """Defensive cycle detection at verify time.""" register_knowtation_identity_provider(repo_root=None) _write_identity( identity_dir, "@org-a", kind="org", members=["@org-b"], quorum_threshold=1, ) _write_identity( identity_dir, "@org-b", kind="org", members=["@org-a"], quorum_threshold=1, ) # No need to even sign — the chain walker rejects before sig checks. commit = _FakeCommit( "c" * 64, {"quorum": json.dumps({ "org_handle": "@org-a", "threshold": 1, "signers": [{"handle": "alice", "public_key": "pk", "signature": "s"}], })}, ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "CYCLE_DETECTED" in result["reason"] def test_provider_missing_is_partial(self, env: dict[str, Any]) -> None: """No identity provider registered → partial=True (NEVER silent True).""" unregister_identity_provider(PROVIDER_DOMAIN) members = [ ("alice", env["members"]["alice"][0], env["members"]["alice"][1]), ("bob", env["members"]["bob"][0], env["members"]["bob"][1]), ] meta_blob = _signed_quorum_metadata("@team", 2, members, env["payload"]) commit = _FakeCommit( env["commit_id"], {"quorum": meta_blob}, author="aaron", agent_id="agent-x", committed_at=env["committed_at"], ) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert result["partial"] is True assert "IDENTITY_PROVIDER_MISSING" in result["reason"] def test_no_quorum_metadata_rejected(self) -> None: """Commit without metadata['quorum'] → reject without provider lookup.""" commit = _FakeCommit("d" * 64, {}) result = verify_quorum_signatures(commit, pathlib.Path("/tmp")) assert result["valid"] is False assert "NO_QUORUM_METADATA" in result["reason"] def test_path_traversal_handle_rejected( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: """Handles with ``..`` / ``/`` MUST raise — never read arbitrary files.""" with pytest.raises(AttestationVerifyError): fs_provider.get_identity("../etc/passwd") with pytest.raises(AttestationVerifyError): fs_provider.get_identity("..\\..\\windows\\system32") def test_oversized_record_rejected( self, identity_dir: pathlib.Path, fs_provider: KnowtationFsIdentityProvider, ) -> None: """Identity files > _MAX_RECORD_BYTES MUST raise.""" big = json.dumps({ "handle": "alice", "kind": "human", "public_key": "x", "padding": "A" * 20_000, }) (identity_dir / "alice.json").write_text(big, encoding="utf-8") with pytest.raises(AttestationVerifyError) as excinfo: fs_provider.get_identity("alice") assert excinfo.value.code == "RECORD_TOO_LARGE" # =========================================================================== # UNIT TIER — registry + concurrency # =========================================================================== class TestRegistryConcurrency: """Identity provider registry is thread-safe under concurrent register/get.""" def test_register_and_get(self, identity_dir: pathlib.Path) -> None: provider = register_knowtation_identity_provider(repo_root=None) from muse.core.attestation import get_identity_provider retrieved = get_identity_provider(PROVIDER_DOMAIN) assert retrieved is provider def test_unregister(self, identity_dir: pathlib.Path) -> None: register_knowtation_identity_provider(repo_root=None) assert unregister_identity_provider(PROVIDER_DOMAIN) is True assert unregister_identity_provider(PROVIDER_DOMAIN) is False def test_concurrent_register_get(self, identity_dir: pathlib.Path) -> None: from muse.core.attestation import get_identity_provider errors: list[Exception] = [] def worker() -> None: try: for _ in range(100): register_knowtation_identity_provider(repo_root=None) get_identity_provider(PROVIDER_DOMAIN) unregister_identity_provider(PROVIDER_DOMAIN) except Exception as e: errors.append(e) threads = [threading.Thread(target=worker) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert errors == [] # =========================================================================== # END-TO-END TIER — muse commit --quorum-signers / verify-commit --verify-identity # =========================================================================== def _write_pem(path: pathlib.Path, priv: Ed25519PrivateKey) -> None: """Write a PEM-encoded private key to *path*.""" pem = priv.private_bytes( encoding=Encoding.PEM, format=PrivateFormat.PKCS8, encryption_algorithm=NoEncryption(), ) path.write_bytes(pem) def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Initialise a knowtation-domain repo at *tmp_path* and chdir into it.""" import argparse as _argparse import os as _os _os.chdir(tmp_path) from muse.cli.commands.init import run as init_run args = _argparse.Namespace( bare=False, template=None, default_branch="main", force=False, domain="knowtation", as_json=False, ) init_run(args) (tmp_path / "notes").mkdir(exist_ok=True) (tmp_path / "notes" / "seed.md").write_text("# seed\n", encoding="utf-8") return tmp_path def _run_commit(tmp_path: pathlib.Path, **overrides: Any) -> dict[str, Any]: """Run muse commit (json mode) with sensible defaults + overrides. Captures stdout AND stderr so failed commits surface their reason via ``AssertionError`` (otherwise the test sees only ``ExitCode.USER_ERROR`` with no diagnostic). """ import argparse as _argparse import io as _io import sys as _sys from muse.cli.commands import commit as commit_cmd base = dict( message="first", allow_empty=False, dry_run=False, section=None, track=None, emotion=None, author=None, agent_id=None, model_id=None, toolchain_id=None, event_type=None, meta=None, sign=False, attest=False, attest_config=None, quorum_signers=None, quorum_org=None, quorum_threshold=None, quorum_keys=None, fmt="json", ) base.update(overrides) out_buf = _io.StringIO() err_buf = _io.StringIO() real_out, real_err = _sys.stdout, _sys.stderr raised: BaseException | None = None try: _sys.stdout, _sys.stderr = out_buf, err_buf try: commit_cmd.run(_argparse.Namespace(**base)) except BaseException as exc: raised = exc finally: _sys.stdout, _sys.stderr = real_out, real_err out_text = out_buf.getvalue() err_text = err_buf.getvalue() if raised is not None: # Attach captured buffers to the exception so the test sees them. raised.captured_stdout = out_text # type: ignore[attr-defined] raised.captured_stderr = err_text # type: ignore[attr-defined] raise raised payload_lines = [ line for line in out_text.splitlines() if line.strip().startswith("{") ] assert payload_lines, ( f"commit produced no JSON output. stdout={out_text!r} stderr={err_text!r}" ) return json.loads(payload_lines[-1]) class TestCommitQuorumE2E: """End-to-end: muse commit --quorum-signers writes the quorum metadata.""" def test_three_member_two_of_three_quorum_round_trip( self, tmp_path: pathlib.Path, identity_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: # Prepare 3 members with PEMs. keys: dict[str, tuple[Ed25519PrivateKey, str]] = {} for h in ("alice", "bob", "carol"): priv, pub = _new_keypair() keys[h] = (priv, pub) pem_path = tmp_path / f"{h}.pem" _write_pem(pem_path, priv) monkeypatch.setenv( f"MUSE_QUORUM_KEY_{h.upper()}_PATH", str(pem_path) ) _write_identity(identity_dir, h, kind="human", public_key=pub) _write_identity( identity_dir, "@team", kind="org", members=list(keys), quorum_threshold=2, ) register_knowtation_identity_provider(repo_root=None) # Run a commit with two co-signers. commit_cmd.run() does NOT # raise SystemExit on the success path (just returns), so the test # also tolerates the non-exit case. repo = _init_repo(tmp_path) result_payload: dict[str, Any] | None = None try: result_payload = _run_commit( repo, message="quorum-test", quorum_signers="alice,bob", quorum_org="@team", quorum_threshold=2, ) except SystemExit as exc: stderr_text = getattr(exc, "captured_stderr", "") stdout_text = getattr(exc, "captured_stdout", "") assert exc.code in (0, None), ( f"commit failed: exit={exc.code}\n" f"stderr={stderr_text!r}\nstdout={stdout_text!r}" ) assert result_payload is None or "commit_id" in result_payload # Read the freshly-written commit and verify the quorum metadata # round-trips through verify_quorum_signatures. from muse.core.store import get_head_commit_id, read_commit head = get_head_commit_id(repo, "main") assert head is not None commit = read_commit(repo, head) assert commit is not None assert "quorum" in commit.metadata result = verify_quorum_signatures(commit, repo) assert result["valid"] is True, result.get("reason") assert result["valid_signers"] == 2 assert result["threshold"] == 2 def test_missing_quorum_key_aborts_commit( self, tmp_path: pathlib.Path, identity_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """Signature stripping defence: missing key → ExitCode.USER_ERROR.""" # Only one member has a key on disk; alice is intentionally missing. bob_priv, bob_pub = _new_keypair() bob_path = tmp_path / "bob.pem" _write_pem(bob_path, bob_priv) monkeypatch.setenv("MUSE_QUORUM_KEY_BOB_PATH", str(bob_path)) _write_identity(identity_dir, "alice", kind="human", public_key="pk-a") _write_identity(identity_dir, "bob", kind="human", public_key=bob_pub) _write_identity( identity_dir, "@team", kind="org", members=["alice", "bob"], quorum_threshold=2, ) # Make sure alice's default ~/.muse/keys path is NOT inadvertently # found — point HOME at the tmp dir to isolate. monkeypatch.setenv("HOME", str(tmp_path)) repo = _init_repo(tmp_path) with pytest.raises(SystemExit) as excinfo: _run_commit( repo, message="missing-key", quorum_signers="alice,bob", quorum_org="@team", quorum_threshold=2, ) # ExitCode.USER_ERROR == 1 assert excinfo.value.code == 1