"""Tests for muse.plugins.knowtation.attestation — Phase 7.2. Rule #0 tier coverage: * unit — id derivation, inbox detection, request-body builder, URL validator, config loader. * integration — register provider, compute → anchor → verify with a mocked endpoint. * security — air.required hard rejection, inbox bypass, malformed URL rejection, placeholder-sig is provably-bad, missing HMAC key is hard error not silent True. * data-integrity — cross-language parity: 200 fixtures produce identical UTF-8 request bodies between Python and air.mjs (Node subprocess; xfail when Node is unavailable). Stress / performance / e2e tiers omitted — this layer is a thin port over ``urllib`` with no async dispatch; the hot-path stress test belongs to Phase 7.4 (the ICP push hook) where 100 concurrent calls actually exercise something interesting. """ from __future__ import annotations import http.server import json import os import shutil import subprocess import threading from typing import Any, cast import pytest from muse.core.attestation import ( AnchorReceipt, AttestationRequiredError, AttestationVerifyError, ImmutableReceiptError, get_attestation_provider, unregister_attestation_provider, ) from muse.plugins.knowtation.attestation import ( ALGORITHM, PLACEHOLDER_SIG, PROVIDER_DOMAIN, AirConfig, KnowtationHmacAttestationProvider, build_air_request_body, canonical_record_bytes, compute_attestation_id, is_inbox_path, load_air_config, register_knowtation_attestation_provider, ) # --------------------------------------------------------------------------- # Cleanup fixture — keep registry pristine. # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def _clean_registry() -> None: """Unregister the knowtation provider after each test.""" yield unregister_attestation_provider(PROVIDER_DOMAIN) @pytest.fixture(autouse=True) def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None: """Strip AIR-related env vars so tests are deterministic.""" monkeypatch.delenv("KNOWTATION_AIR_ENDPOINT", raising=False) monkeypatch.delenv("MUSE_AIR_HMAC_KEY", raising=False) monkeypatch.delenv("MUSE_AIR_CONFIG", raising=False) # =========================================================================== # UNIT TIER # =========================================================================== class TestInboxDetection: """Exact parity with isInboxPath() in air.mjs.""" @pytest.mark.parametrize( "path,expected", [ ("inbox", True), ("inbox/foo.md", True), ("inbox/sub/bar.md", True), ("projects/alpha/inbox", True), ("projects/alpha/inbox/note.md", True), ("projects/alpha-beta/inbox/x", True), ("notes/inbox.md", False), ("projects/inbox/note.md", False), ("projects/alpha/sub/inbox/note.md", False), ("inbox-archive/x", False), ("", False), ], ) def test_classification(self, path: str, expected: bool) -> None: """Direct verbatim parity with the JS regex.""" assert is_inbox_path(path) is expected def test_backslash_normalisation(self) -> None: """Windows paths are normalised before matching.""" assert is_inbox_path("inbox\\foo.md") is True assert is_inbox_path("projects\\alpha\\inbox\\x") is True class TestIdDerivation: """compute_attestation_id is deterministic and unique.""" def test_deterministic(self) -> None: """Same inputs → same id, every call.""" a = compute_attestation_id("snap-1", "notes/x.md") b = compute_attestation_id("snap-1", "notes/x.md") assert a == b assert len(a) == 64 assert a.islower() def test_distinct_for_different_inputs(self) -> None: """Distinct (snapshot,path) pairs collide with negligible probability.""" ids = { compute_attestation_id("snap-1", "notes/x.md"), compute_attestation_id("snap-2", "notes/x.md"), compute_attestation_id("snap-1", "notes/y.md"), } assert len(ids) == 3 def test_separator_injection_resistance(self) -> None: """Null-byte separator means concatenation aliases cannot collide.""" with pytest.raises(ValueError): compute_attestation_id("snap\x00inj", "notes/x.md") def test_rejects_non_strings(self) -> None: """Non-string args are rejected at the boundary.""" with pytest.raises(ValueError): compute_attestation_id(cast(str, 1), "x") # type: ignore[arg-type] class TestRequestBodyBuilder: """build_air_request_body produces canonical JSON.""" def test_keys_in_alphabetical_order(self) -> None: """JSON keys appear in canonical alphabetical order.""" body = build_air_request_body("write", "notes/x.md", "abc") # Decoded text order is what air.mjs (and any strict JSON consumer) # observes; assert the *string* form keeps the canonical order. text = body.decode("utf-8") assert text == '{"action":"write","content_hash":"abc","path":"notes/x.md"}' def test_unicode_path_passthrough(self) -> None: """Unicode paths survive round-trip without escaping.""" body = build_air_request_body("write", "notes/résumé.md", "abc") decoded = json.loads(body) assert decoded == {"action": "write", "content_hash": "abc", "path": "notes/résumé.md"} def test_rejects_null_byte(self) -> None: """Null-byte values would break canonical encoding — rejected.""" with pytest.raises(ValueError, match="null"): build_air_request_body("write", "x\x00y", "abc") def test_rejects_non_string(self) -> None: """Non-string args are rejected at the boundary.""" with pytest.raises(ValueError): build_air_request_body(cast(str, 1), "x", "abc") # type: ignore[arg-type] class TestCanonicalRecordBytes: """canonical_record_bytes is the input to HMAC sign/verify.""" def test_includes_required_fields(self) -> None: """All five canonical fields appear, null-byte separated, correct order.""" record = { "id": "abc" * 21 + "x", "action": "write", "path": "notes/x.md", "content_hash": "deadbeef", "timestamp": "2026-01-01T00:00:00Z", "sig": "should-not-appear", "algorithm": ALGORITHM, "provider_id": PROVIDER_DOMAIN, } out = canonical_record_bytes(cast(Any, record)) decoded = out.decode("utf-8") parts = decoded.split("\x00") assert parts == ["write", "notes/x.md", "deadbeef", "2026-01-01T00:00:00Z", record["id"]] assert "should-not-appear" not in decoded # =========================================================================== # INTEGRATION TIER # =========================================================================== class _LocalAir: """Tiny in-process HTTP server used by integration tests.""" def __init__(self, response: dict[str, Any] | None = None, status: int = 200) -> None: self.response = response or {"id": "ignored", "air_sig": "deadbeef-sig"} self.status = status self.requests: list[bytes] = [] outer = self class _Handler(http.server.BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 — stdlib name length = int(self.headers.get("Content-Length", "0")) outer.requests.append(self.rfile.read(length)) self.send_response(outer.status) self.send_header("Content-Type", "application/json") body = json.dumps(outer.response).encode("utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, fmt: str, *args: Any) -> None: # noqa: ARG002 return None self._server = http.server.HTTPServer(("127.0.0.1", 0), _Handler) self.endpoint = f"http://127.0.0.1:{self._server.server_address[1]}" self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._thread.start() def stop(self) -> None: self._server.shutdown() self._server.server_close() self._thread.join(timeout=2.0) @pytest.fixture def local_air() -> Any: """Local AIR endpoint that returns a real signature.""" server = _LocalAir() yield server server.stop() class TestRoundTrip: """register → compute → anchor → verify with a real local endpoint.""" def test_compute_anchor_returns_valid_record(self, local_air: _LocalAir) -> None: """Happy path: configured endpoint yields a non-placeholder record.""" cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} provider = register_knowtation_attestation_provider(cfg) record = provider.compute_attestation( "snap-abc", {"path": "notes/x.md", "content_hash": "deadbeef"} ) assert record["id"] == compute_attestation_id("snap-abc", "notes/x.md") assert record["sig"] == "deadbeef-sig" assert record["algorithm"] == ALGORITHM assert record["provider_id"] == PROVIDER_DOMAIN receipt = provider.anchor(record) assert receipt["tx_id"] == f"air-local:{record['id']}" assert receipt["provider_id"] == PROVIDER_DOMAIN def test_request_body_is_canonical(self, local_air: _LocalAir) -> None: """Endpoint receives exactly the canonical body shape.""" cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} provider = KnowtationHmacAttestationProvider(cfg) provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) assert local_air.requests body = local_air.requests[0].decode("utf-8") assert body == '{"action":"write","content_hash":"abc","path":"notes/x.md"}' def test_get_via_registry(self) -> None: """register_knowtation_attestation_provider plumbs into the registry.""" provider = register_knowtation_attestation_provider({"enabled": False}) assert get_attestation_provider(PROVIDER_DOMAIN) is provider class TestVerifyWithKey: """Round-trip verify when both sides know the HMAC key.""" def test_local_hmac_verify_round_trip(self, monkeypatch: pytest.MonkeyPatch) -> None: """Sign with key → verify with same key → valid=True.""" # Use a known key, sign locally, then verify. key_hex = "00" * 32 provider = KnowtationHmacAttestationProvider({"hmac_key": key_hex}) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) # Build a synthetic record with a real HMAC sig so verify succeeds. from muse.plugins.knowtation.attestation import ( _hmac_sha256_hex, canonical_record_bytes, ) record["timestamp"] = "2026-01-01T00:00:00Z" sig = _hmac_sha256_hex(key_hex, canonical_record_bytes(record)) record["sig"] = sig record["algorithm"] = ALGORITHM receipt = provider.anchor(record) result = provider.verify(record, receipt) assert result["valid"] is True assert result["depth"] == 0 def test_tampered_record_returns_valid_false(self, monkeypatch: pytest.MonkeyPatch) -> None: """Mutating any HMAC-bound field after signing flips valid to False.""" key_hex = "00" * 32 provider = KnowtationHmacAttestationProvider({"hmac_key": key_hex}) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) from muse.plugins.knowtation.attestation import ( _hmac_sha256_hex, canonical_record_bytes, ) record["timestamp"] = "2026-01-01T00:00:00Z" sig = _hmac_sha256_hex(key_hex, canonical_record_bytes(record)) record["sig"] = sig receipt = provider.anchor(record) record["content_hash"] = "tampered-hash" result = provider.verify(record, receipt) assert result["valid"] is False assert "HMAC" in result["reason"] # =========================================================================== # SECURITY TIER (mandatory) # =========================================================================== class TestSecurity: """Hardening contracts — Rule #0 mandates this tier for Phase 7.""" def test_inbox_bypass_skips_endpoint(self, local_air: _LocalAir) -> None: """Inbox paths never call the AIR endpoint.""" cfg: AirConfig = {"enabled": True, "required": True, "endpoint": local_air.endpoint} provider = KnowtationHmacAttestationProvider(cfg) record = provider.compute_attestation( "snap", {"path": "inbox/note.md", "content_hash": "x"} ) assert record["algorithm"] == "bypass-inbox" assert record["sig"] == "" assert local_air.requests == [] def test_required_true_endpoint_500_raises(self) -> None: """500 from endpoint with required=True → AttestationRequiredError.""" server = _LocalAir(status=500) try: cfg: AirConfig = {"enabled": True, "required": True, "endpoint": server.endpoint} provider = KnowtationHmacAttestationProvider(cfg) with pytest.raises(AttestationRequiredError): provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) finally: server.stop() def test_required_false_endpoint_500_returns_placeholder(self) -> None: """500 from endpoint with required=False → placeholder, no raise.""" server = _LocalAir(status=500) try: cfg: AirConfig = {"enabled": True, "required": False, "endpoint": server.endpoint} provider = KnowtationHmacAttestationProvider(cfg) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) assert record["sig"] == PLACEHOLDER_SIG assert record["algorithm"].startswith("placeholder-") finally: server.stop() def test_required_true_no_endpoint_raises(self) -> None: """No endpoint configured + required=True → AttestationRequiredError.""" cfg: AirConfig = {"enabled": True, "required": True} provider = KnowtationHmacAttestationProvider(cfg) with pytest.raises(AttestationRequiredError): provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) def test_malformed_endpoint_url_raises(self) -> None: """file:// and other non-http(s) schemes are rejected — no SSRF.""" cfg: AirConfig = {"enabled": True, "required": True, "endpoint": "file:///etc/passwd"} provider = KnowtationHmacAttestationProvider(cfg) with pytest.raises(AttestationRequiredError): provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) def test_anchor_is_write_once(self, local_air: _LocalAir) -> None: """Re-anchoring the same record raises ImmutableReceiptError.""" cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} provider = KnowtationHmacAttestationProvider(cfg) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) provider.anchor(record) with pytest.raises(ImmutableReceiptError): provider.anchor(record) def test_verify_placeholder_returns_valid_false_not_true(self) -> None: """Placeholder sig is provably-bad: valid=False, never silent True.""" provider = KnowtationHmacAttestationProvider({}) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) receipt = provider.anchor(record) result = provider.verify(record, receipt) assert result["valid"] is False assert "placeholder" in result["reason"].lower() def test_verify_missing_hmac_key_raises_not_returns( self, monkeypatch: pytest.MonkeyPatch, local_air: _LocalAir ) -> None: """Missing HMAC key → AttestationVerifyError, never silent True.""" cfg: AirConfig = {"enabled": True, "endpoint": local_air.endpoint} provider = KnowtationHmacAttestationProvider(cfg) record = provider.compute_attestation( "snap", {"path": "notes/x.md", "content_hash": "abc"} ) receipt = provider.anchor(record) with pytest.raises(AttestationVerifyError) as exc: provider.verify(record, receipt) assert exc.value.code == "KEY_MISSING" def test_verify_inbox_record_returns_valid_false(self) -> None: """Inbox-bypass records cannot be verified — valid=False, never True.""" provider = KnowtationHmacAttestationProvider({"enabled": True}) record = provider.compute_attestation( "snap", {"path": "inbox/x.md", "content_hash": "abc"} ) receipt = provider.anchor(record) result = provider.verify(record, receipt) assert result["valid"] is False def test_verify_provider_id_mismatch_raises(self) -> None: """Records from a different provider raise, never return valid=True.""" provider = KnowtationHmacAttestationProvider({"hmac_key": "00" * 32}) evil_record: Any = { "id": "a" * 64, "action": "write", "path": "notes/x.md", "content_hash": "x", "timestamp": "2026-01-01T00:00:00Z", "sig": "any", "algorithm": ALGORITHM, "provider_id": "evil-domain", } evil_receipt: Any = { "tx_id": "x", "anchor_url": "https://evil.example/x", "anchored_at": "2026-01-01T00:00:00Z", "provider_id": "evil-domain", } with pytest.raises(AttestationVerifyError) as exc: provider.verify(evil_record, evil_receipt) assert exc.value.code == "PROVIDER_MISMATCH" # =========================================================================== # DATA-INTEGRITY TIER — cross-language parity with air.mjs # =========================================================================== _NODE = shutil.which("node") _AIR_MJS = "/Users/aaronrenecarvajal/knowtation/lib/air.mjs" def _node_build_body(action: str, path: str, content_hash: str) -> bytes: """Invoke node to call buildAirRequestBody — returns the same bytes Python should.""" script = ( f"import('{_AIR_MJS}').then(m => " "{ " f"const out = m.buildAirRequestBody({json.dumps(action)}, " f"{json.dumps(path)}, {json.dumps(content_hash)}); " "process.stdout.write(out); " "}).catch(e => { console.error(e); process.exit(2); });" ) proc = subprocess.run( [_NODE, "--input-type=module", "-e", script], capture_output=True, timeout=10, check=False, ) if proc.returncode != 0: raise RuntimeError(f"node failed: {proc.stderr.decode('utf-8', errors='replace')}") return proc.stdout @pytest.mark.skipif( _NODE is None or not os.path.exists(_AIR_MJS), reason="Node or air.mjs unavailable — cross-language parity test gated", ) class TestCrossLanguageParity: """200 fixture inputs produce identical Python and JS request bodies.""" def _fixtures(self) -> list[tuple[str, str, str]]: """Generate 200 deterministic (action, path, content_hash) fixtures.""" actions = ["write", "export", "import", "consolidation"] paths = [ "notes/a.md", "projects/alpha/note.md", "deep/nested/path/with-dashes_underscores.md", "unicode/résumé.md", "spaces in name.md", "", "single", ] out: list[tuple[str, str, str]] = [] i = 0 while len(out) < 200: action = actions[i % len(actions)] path = paths[(i // len(actions)) % len(paths)] ch = f"{i:064x}" out.append((action, path, ch)) i += 1 return out def test_200_fixtures_byte_identical(self) -> None: """For every fixture, Python and JS produce the same UTF-8 bytes.""" mismatches: list[tuple[tuple[str, str, str], bytes, bytes]] = [] for fix in self._fixtures(): py = build_air_request_body(*fix) js = _node_build_body(*fix) if py != js: mismatches.append((fix, py, js)) assert not mismatches, ( f"{len(mismatches)} fixtures produced different bytes; " f"first: {mismatches[0]!r}" ) # =========================================================================== # CONFIG LOADER # =========================================================================== class TestConfigLoader: """load_air_config reads the YAML and respects env-var fallbacks.""" def test_missing_file_returns_empty(self, tmp_path: Any) -> None: """Missing file → empty AirConfig, no exception.""" cfg = load_air_config(tmp_path / "nope.yaml") assert cfg == {} def test_yaml_parsed(self, tmp_path: Any) -> None: """Valid YAML is parsed into the AirConfig fields.""" path = tmp_path / "air.yaml" path.write_text( "air:\n enabled: true\n required: false\n endpoint: https://x.example/air\n", encoding="utf-8", ) cfg = load_air_config(path) assert cfg["enabled"] is True assert cfg["required"] is False assert cfg["endpoint"] == "https://x.example/air" def test_env_var_path_used(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Any) -> None: """MUSE_AIR_CONFIG env var overrides the default path.""" path = tmp_path / "custom.yaml" path.write_text("air:\n enabled: true\n", encoding="utf-8") monkeypatch.setenv("MUSE_AIR_CONFIG", str(path)) cfg = load_air_config() assert cfg["enabled"] is True def test_malformed_root_raises(self, tmp_path: Any) -> None: """Top-level YAML not a mapping → AttestationError.""" from muse.core.attestation import AttestationError path = tmp_path / "bad.yaml" path.write_text("- a\n- b\n", encoding="utf-8") with pytest.raises(AttestationError): load_air_config(path)