test_phase7_exit_criteria.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
26 days ago
| 1 | """Phase 7 exit-criteria smoke test. |
| 2 | |
| 3 | Exercises the whole Phase 7 surface end-to-end on a synthetic vault: |
| 4 | |
| 5 | 1. ``muse commit --sign --attest --meta '{...}' --quorum-signers a,b --quorum-org @team`` |
| 6 | 2. ``muse verify-commit --attest --verify-identity`` → tier 2 (HMAC; ICP gated by |
| 7 | the network-isolated test environment) AND ``quorum_valid=True``. |
| 8 | |
| 9 | Real network calls are intentionally avoided — the ICP canister query is |
| 10 | disabled via ``--no-icp`` so the test deterministically stops at tier 2 and |
| 11 | exercises the full software path WITHOUT a live IC dependency. |
| 12 | |
| 13 | Why this test exists: |
| 14 | The user's Phase 7 spec ends with:: |
| 15 | |
| 16 | Phase 7 exit criteria |
| 17 | --------------------- |
| 18 | `muse commit --sign --attest --meta '{...}' ` produces a commit verifiable |
| 19 | via `muse verify-commit --attest --verify-identity`; ICP anchor visible |
| 20 | on MuseHub note detail page; org-quorum commits sign and verify |
| 21 | end-to-end on a fixture vault. |
| 22 | |
| 23 | This file proves the CLI surface meets that contract on a fixture vault. |
| 24 | The MuseHub note-detail page rendering is covered separately by |
| 25 | ``test_musehub_provenance_badge_strip.py``. |
| 26 | """ |
| 27 | |
| 28 | from __future__ import annotations |
| 29 | |
| 30 | import argparse |
| 31 | import io |
| 32 | import json |
| 33 | import os |
| 34 | import pathlib |
| 35 | import sys |
| 36 | from typing import Any |
| 37 | |
| 38 | import pytest |
| 39 | from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey |
| 40 | from cryptography.hazmat.primitives.serialization import ( |
| 41 | Encoding, NoEncryption, PrivateFormat, |
| 42 | ) |
| 43 | |
| 44 | from muse.core.attestation import ( |
| 45 | AnchorReceipt, |
| 46 | AttestationRecord, |
| 47 | MuseAttestationProvider, |
| 48 | VerifyResult, |
| 49 | register_attestation_provider, |
| 50 | unregister_attestation_provider, |
| 51 | unregister_identity_provider, |
| 52 | ) |
| 53 | from muse.core.provenance import encode_public_key |
| 54 | from muse.plugins.knowtation.quorum import ( |
| 55 | PROVIDER_DOMAIN as IDENTITY_DOMAIN, |
| 56 | register_knowtation_identity_provider, |
| 57 | ) |
| 58 | |
| 59 | |
| 60 | # Same write-once stub that the existing 7.3 / 7.5 suites use. |
| 61 | _HMAC_DOMAIN = "knowtation" |
| 62 | |
| 63 | |
| 64 | class _StubAttestProvider: |
| 65 | """Always-valid attestation provider used for the smoke test.""" |
| 66 | |
| 67 | def __init__(self) -> None: |
| 68 | self.calls: list[tuple[str, dict[str, object]]] = [] |
| 69 | |
| 70 | def compute_attestation( |
| 71 | self, snapshot_id: str, commit_meta: dict[str, object] |
| 72 | ) -> AttestationRecord: |
| 73 | self.calls.append((snapshot_id, dict(commit_meta))) |
| 74 | return { |
| 75 | "id": "f" * 64, |
| 76 | "action": "write", |
| 77 | "path": str(commit_meta.get("path", "")), |
| 78 | "timestamp": str(commit_meta.get("timestamp", "")), |
| 79 | "content_hash": snapshot_id, |
| 80 | "sig": "smoke-sig", |
| 81 | "algorithm": "smoke", |
| 82 | "provider_id": _HMAC_DOMAIN, |
| 83 | } |
| 84 | |
| 85 | def anchor(self, record: AttestationRecord) -> AnchorReceipt: |
| 86 | return { |
| 87 | "tx_id": "smoke-tx", "anchor_url": "", |
| 88 | "anchored_at": "", "provider_id": _HMAC_DOMAIN, |
| 89 | } |
| 90 | |
| 91 | def verify(self, record: AttestationRecord, receipt: AnchorReceipt) -> VerifyResult: |
| 92 | return {"valid": True, "reason": "", "depth": 0, "provider_id": _HMAC_DOMAIN} |
| 93 | |
| 94 | |
| 95 | def _new_key_pem(path: pathlib.Path) -> tuple[Ed25519PrivateKey, str]: |
| 96 | """Generate an Ed25519 keypair, write the PEM, return (priv, pub_b64).""" |
| 97 | priv = Ed25519PrivateKey.generate() |
| 98 | _raw, pub_b64 = encode_public_key(priv) |
| 99 | pem = priv.private_bytes( |
| 100 | encoding=Encoding.PEM, |
| 101 | format=PrivateFormat.PKCS8, |
| 102 | encryption_algorithm=NoEncryption(), |
| 103 | ) |
| 104 | path.write_bytes(pem) |
| 105 | return priv, pub_b64 |
| 106 | |
| 107 | |
| 108 | def _write_identity( |
| 109 | dir_: pathlib.Path, |
| 110 | handle: str, |
| 111 | *, |
| 112 | kind: str, |
| 113 | public_key: str = "", |
| 114 | members: list[str] | None = None, |
| 115 | quorum_threshold: int = 0, |
| 116 | ) -> None: |
| 117 | dir_.mkdir(parents=True, exist_ok=True) |
| 118 | fname = ("_at_" + handle[1:]) if handle.startswith("@") else handle |
| 119 | body: dict[str, Any] = { |
| 120 | "handle": handle, "kind": kind, "public_key": public_key, |
| 121 | } |
| 122 | if members is not None: |
| 123 | body["members"] = members |
| 124 | if quorum_threshold: |
| 125 | body["quorum_threshold"] = quorum_threshold |
| 126 | (dir_ / f"{fname}.json").write_text(json.dumps(body), encoding="utf-8") |
| 127 | |
| 128 | |
| 129 | def _capture_run(fn: Any, *args: Any, **kwargs: Any) -> tuple[int | None, str, str]: |
| 130 | """Run *fn* capturing stdout/stderr; tolerate SystemExit.""" |
| 131 | out_buf = io.StringIO() |
| 132 | err_buf = io.StringIO() |
| 133 | real_out, real_err = sys.stdout, sys.stderr |
| 134 | code: int | None = None |
| 135 | try: |
| 136 | sys.stdout, sys.stderr = out_buf, err_buf |
| 137 | try: |
| 138 | fn(*args, **kwargs) |
| 139 | except SystemExit as exc: |
| 140 | code = ( |
| 141 | exc.code if isinstance(exc.code, int) |
| 142 | else (0 if exc.code is None else 1) |
| 143 | ) |
| 144 | finally: |
| 145 | sys.stdout, sys.stderr = real_out, real_err |
| 146 | return code, out_buf.getvalue(), err_buf.getvalue() |
| 147 | |
| 148 | |
| 149 | @pytest.fixture(autouse=True) |
| 150 | def _registry_cleanup() -> Any: |
| 151 | unregister_attestation_provider(_HMAC_DOMAIN) |
| 152 | unregister_identity_provider(IDENTITY_DOMAIN) |
| 153 | yield |
| 154 | unregister_attestation_provider(_HMAC_DOMAIN) |
| 155 | unregister_identity_provider(IDENTITY_DOMAIN) |
| 156 | |
| 157 | |
| 158 | def test_full_phase7_round_trip( |
| 159 | tmp_path: pathlib.Path, |
| 160 | monkeypatch: pytest.MonkeyPatch, |
| 161 | ) -> None: |
| 162 | """commit --sign --attest --meta --quorum → verify-commit verifies all tiers.""" |
| 163 | # ── 1. Fixture vault setup ─────────────────────────────────────────── |
| 164 | os.chdir(tmp_path) |
| 165 | from muse.cli.commands.init import run as init_run |
| 166 | |
| 167 | init_run(argparse.Namespace( |
| 168 | bare=False, template=None, default_branch="main", |
| 169 | force=False, domain="knowtation", as_json=False, |
| 170 | )) |
| 171 | (tmp_path / "notes").mkdir(exist_ok=True) |
| 172 | (tmp_path / "notes" / "smoke.md").write_text( |
| 173 | "---\ntitle: Smoke\n---\n# smoke\n", encoding="utf-8", |
| 174 | ) |
| 175 | |
| 176 | # ── 2. Attestation provider + identity directory ──────────────────── |
| 177 | register_attestation_provider(_HMAC_DOMAIN, _StubAttestProvider()) |
| 178 | identity_dir = tmp_path / "identity-cache" |
| 179 | monkeypatch.setenv("MUSE_IDENTITY_DIR", str(identity_dir)) |
| 180 | register_knowtation_identity_provider(repo_root=None) |
| 181 | |
| 182 | # Primary signing key — injected via MUSE_AGENT_KEY so the engine's |
| 183 | # offline path picks it up without a registered hub. |
| 184 | primary_pem_path = tmp_path / "primary.pem" |
| 185 | _new_key_pem(primary_pem_path) |
| 186 | monkeypatch.setenv("MUSE_AGENT_KEY", primary_pem_path.read_text()) |
| 187 | monkeypatch.setenv("MUSE_AGENT_HANDLE", "smoke-agent") |
| 188 | |
| 189 | # Two co-signers + the org record they're members of. |
| 190 | alice_priv, alice_pub = _new_key_pem(tmp_path / "alice.pem") |
| 191 | bob_priv, bob_pub = _new_key_pem(tmp_path / "bob.pem") |
| 192 | monkeypatch.setenv("MUSE_QUORUM_KEY_ALICE_PATH", str(tmp_path / "alice.pem")) |
| 193 | monkeypatch.setenv("MUSE_QUORUM_KEY_BOB_PATH", str(tmp_path / "bob.pem")) |
| 194 | _write_identity(identity_dir, "alice", kind="human", public_key=alice_pub) |
| 195 | _write_identity(identity_dir, "bob", kind="human", public_key=bob_pub) |
| 196 | _write_identity( |
| 197 | identity_dir, "@team", kind="org", |
| 198 | members=["alice", "bob"], quorum_threshold=2, |
| 199 | ) |
| 200 | |
| 201 | # ── 3. muse commit --sign --attest --meta --quorum-signers ────────── |
| 202 | from muse.cli.commands import commit as commit_cmd |
| 203 | |
| 204 | args = argparse.Namespace( |
| 205 | message="phase7-smoke", allow_empty=False, dry_run=False, |
| 206 | section=None, track=None, emotion=None, author=None, |
| 207 | agent_id="smoke-agent", model_id=None, toolchain_id=None, |
| 208 | event_type=None, |
| 209 | meta=json.dumps({"phase": "7-smoke", "topic": "exit-criteria"}), |
| 210 | sign=True, # primary key resolved from MUSE_AGENT_KEY env var. |
| 211 | attest=True, attest_config=None, |
| 212 | quorum_signers="alice,bob", quorum_org="@team", |
| 213 | quorum_threshold=2, quorum_keys=None, |
| 214 | fmt="json", |
| 215 | ) |
| 216 | code, out, err = _capture_run(commit_cmd.run, args) |
| 217 | assert code in (0, None), f"commit failed exit={code}\nstdout={out}\nstderr={err}" |
| 218 | |
| 219 | # ── 4. Read back the commit and confirm metadata wiring ──────────── |
| 220 | from muse.core.store import get_head_commit_id, read_commit |
| 221 | head = get_head_commit_id(tmp_path, "main") |
| 222 | assert head is not None |
| 223 | commit = read_commit(tmp_path, head) |
| 224 | assert commit is not None, "head commit unreadable" |
| 225 | assert "attestation" in commit.metadata, "commit missing attestation record" |
| 226 | assert "quorum" in commit.metadata, "commit missing quorum metadata" |
| 227 | assert "meta" in commit.metadata, "commit missing --meta payload" |
| 228 | meta_payload = json.loads(commit.metadata["meta"]) |
| 229 | assert meta_payload == {"phase": "7-smoke", "topic": "exit-criteria"} |
| 230 | |
| 231 | # ── 5. muse verify-commit --attest --verify-identity --no-icp ────── |
| 232 | # --no-icp keeps the test offline; tier caps at HMAC_ATTESTED (2). |
| 233 | from muse.cli.commands import verify_commit as verify_cmd |
| 234 | |
| 235 | vargs = argparse.Namespace( |
| 236 | commit_id=head, attest=True, no_icp=True, |
| 237 | icp_canister=None, icp_timeout=5.0, |
| 238 | json_out=True, verify_identity=True, identity_domain="knowtation", |
| 239 | ) |
| 240 | vcode, vout, verr = _capture_run(verify_cmd.run, vargs) |
| 241 | assert vout, f"verify-commit produced no JSON output. stderr={verr}" |
| 242 | payload = json.loads(vout) |
| 243 | |
| 244 | # Tier 2 = signed (Ed25519) + HMAC-attested. Tier 3 (ICP-anchored) is |
| 245 | # gated by --no-icp; the canister query is intentionally skipped so this |
| 246 | # smoke test stays offline. |
| 247 | assert payload["tier"] == 2, ( |
| 248 | f"expected HMAC-attested tier; got {payload}" |
| 249 | ) |
| 250 | assert payload["valid"] is True |
| 251 | assert payload["hmac_valid"] is True |
| 252 | quorum = payload["quorum"] |
| 253 | assert quorum["valid"] is True, f"quorum verification failed: {quorum}" |
| 254 | assert quorum["valid_signers"] == 2 |
| 255 | assert quorum["threshold"] == 2 |
| 256 | assert quorum["org_handle"] == "@team" |
| 257 | assert quorum["depth"] >= 2 # @team → alice + bob = depth 2 |
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