"""TDD — Server-side Ed25519 commit signature verification. Security requirement: when a commit arrives with a non-empty ``signature`` AND a non-empty ``signer_public_key``, the server MUST cryptographically verify the Ed25519 signature before accepting the push. A presence-only check is not sufficient — a pusher who passes MSign auth could forge any signature string and the server would store it as verified provenance. Test IDs -------- SV1 Valid signature → push accepted SV2 Forged (garbage) signature bytes → push rejected 422 SV3 Valid signature but wrong public key declared → push rejected 422 SV4 signature present, signer_public_key empty → push rejected 422 SV5 Unsigned commit (require_signed_commits=False) → accepted (regression guard) SV6 Unsigned commit (require_signed_commits=True) → rejected (regression guard) SV7 Mixed batch: one valid + one forged → entire push rejected """ from __future__ import annotations from datetime import datetime, timezone from unittest.mock import AsyncMock import msgpack import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.mpack import WIRE_CONTENT_TYPE from muse.core.provenance import ( encode_public_key, provenance_payload, sign_commit_ed25519, sign_commit_record, ) from muse.core.types import encode_pubkey, now_utc_iso, public_key_fingerprint, blob_id from musehub.db.musehub_models import MusehubBranch, MusehubRepo from musehub.core.genesis import compute_branch_id, compute_identity_id, compute_repo_id from musehub.models.wire import ( SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_ERROR, SFRAME_HEADER, SFRAME_RESULT, ) from muse.core.mpack import MuseWireFrameWriter from musehub.types.json_types import JSONObject, StrDict _fw = MuseWireFrameWriter() # --------------------------------------------------------------------------- # Frame helpers (mirrors test_wire_push_stream.py conventions) # --------------------------------------------------------------------------- def _pack(data: JSONObject) -> bytes: return msgpack.packb(data, use_bin_type=True) def _wrap(ft: str, data: JSONObject) -> bytes: return _fw.wrap(frame_type=ft, payload=_pack(data)) def _header_frame(*, branch: str = "main", n_objects: int = 0, n_commits: int = 0) -> bytes: return _wrap(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": branch, "force": False, "have": [], "n_objects": n_objects, "n_commits": n_commits, }) def _commit_pack_frame(commits: list[dict], snapshots: list[dict] | None = None) -> bytes: return _wrap(SFRAME_COMMIT_PACK, { "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots or [], "snapshot_deltas": [], }) def _end_frame(n_objects: int = 0, n_commits: int = 0) -> bytes: return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits}) # --------------------------------------------------------------------------- # Ed25519 test key generation # --------------------------------------------------------------------------- def _new_keypair() -> tuple[Ed25519PrivateKey, bytes, str]: """Return (private_key, raw_pub_bytes, encoded_pub_str).""" private_key = Ed25519PrivateKey.generate() raw_bytes, pub_str = encode_public_key(private_key) return private_key, raw_bytes, pub_str # --------------------------------------------------------------------------- # Commit builder with optional real Ed25519 signature # --------------------------------------------------------------------------- def _make_commit( *, commit_id: str | None = None, snapshot_id: str | None = None, branch: str = "main", author: str = "gabriel", committed_at: str | None = None, private_key: Ed25519PrivateKey | None = None, # Override specific wire fields after signing: override_signature: str | None = None, override_signer_public_key: str | None = None, override_signer_key_id: str | None = None, ) -> JSONObject: ts = committed_at or now_utc_iso() cid = commit_id or blob_id(f"sv-commit-{ts}".encode()) snap_id = snapshot_id or blob_id(b"sv-default-snap") commit: JSONObject = { "commit_id": cid, "parent_commit_id": None, "parent2_commit_id": None, "snapshot_id": snap_id, "branch": branch, "message": "test commit", "author": author, "committed_at": ts, "signature": "", "signer_public_key": "", "signer_key_id": "", "agent_id": "claude-code", "model_id": "claude-sonnet-4-6", "metadata": {}, } if private_key is not None: result = sign_commit_record( cid, "claude-code", private_key, author=author, model_id="claude-sonnet-4-6", committed_at=ts, ) assert result is not None sig, pub_b64, key_id = result commit["signature"] = sig commit["signer_public_key"] = pub_b64 commit["signer_key_id"] = key_id # Overrides for adversarial tests if override_signature is not None: commit["signature"] = override_signature if override_signer_public_key is not None: commit["signer_public_key"] = override_signer_public_key if override_signer_key_id is not None: commit["signer_key_id"] = override_signer_key_id return commit def _make_snapshot(snapshot_id: str) -> JSONObject: return {"snapshot_id": snapshot_id, "manifest": {}, "committed_at": now_utc_iso()} # --------------------------------------------------------------------------- # DB + backend helpers # --------------------------------------------------------------------------- async def _make_repo( db_session: AsyncSession, name: str, owner: str = "testuser", ) -> MusehubRepo: owner_user_id = compute_identity_id(owner.encode()) slug = name.lower().replace(" ", "-") created_at = datetime.now(tz=timezone.utc) repo_id = compute_repo_id(owner_user_id, slug, "code", created_at.isoformat()) repo = MusehubRepo( repo_id=repo_id, name=name, owner=owner, slug=slug, visibility="public", owner_user_id=owner_user_id, description="", tags=[], created_at=created_at, ) db_session.add(repo) await db_session.commit() branch = MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", ) db_session.add(branch) await db_session.commit() await db_session.refresh(repo) return repo def _stub_r2(monkeypatch: pytest.MonkeyPatch) -> None: _store: dict[str, bytes] = {} backend = AsyncMock() backend.exists = AsyncMock(side_effect=lambda oid: oid in _store) backend.put = AsyncMock(side_effect=lambda oid, data: _store.update({oid: data}) or f"r2://{oid}") backend.get = AsyncMock(side_effect=lambda oid: _store.get(oid)) monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) def _decode_result(resp_content: bytes) -> JSONObject: unpacker = msgpack.Unpacker(raw=False) unpacker.feed(resp_content) last: JSONObject = {} for frame in unpacker: last = frame return last # --------------------------------------------------------------------------- # SV1 — Valid signature: push accepted # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv1_valid_signature_accepted( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV1: A commit with a real Ed25519 signature is accepted by the server.""" _stub_r2(monkeypatch) repo = await _make_repo(db_session, "SV1 Valid Sig", owner="testuser") private_key, _, _ = _new_keypair() snap_id = blob_id(b"sv1-snap") commit = _make_commit(snapshot_id=snap_id, private_key=private_key) snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) assert resp.status_code == 200, f"expected 200, got {resp.status_code}: {result}" assert result.get("ok") is True, f"expected ok=True, got: {result}" # --------------------------------------------------------------------------- # SV2 — Forged/garbage signature: push rejected # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv2_forged_signature_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV2: A commit with a present-but-invalid signature is rejected with 422. The server must not accept a push where signature bytes do not verify against the declared public key, even when require_signed_commits is off. """ _stub_r2(monkeypatch) repo = await _make_repo(db_session, "SV2 Forged Sig", owner="testuser") # Generate a real key so the public key is valid — but forge the signature private_key, raw_pub, pub_str = _new_keypair() key_id = public_key_fingerprint(raw_pub) forged_sig = "ed25519:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" snap_id = blob_id(b"sv2-snap") commit = _make_commit( snapshot_id=snap_id, override_signature=forged_sig, override_signer_public_key=pub_str, override_signer_key_id=key_id, ) snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) # Must be rejected — not silently accepted is_error = result.get("t") == SFRAME_ERROR is_http_error = resp.status_code == 422 assert is_error or is_http_error, ( f"SV2: forged signature must be rejected. " f"Got status={resp.status_code}, result={result}" ) if is_error: assert result.get("code") == 422, f"expected error code 422, got: {result}" # --------------------------------------------------------------------------- # SV3 — Valid signature but wrong declared public key: rejected # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv3_wrong_public_key_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV3: A commit signed with key A but declaring key B is rejected. This catches the impersonation attack where a pusher signs with their own key but claims a trusted agent's public key. """ _stub_r2(monkeypatch) repo = await _make_repo(db_session, "SV3 Wrong Pubkey", owner="testuser") # Sign with key_a, but declare key_b from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat key_a = Ed25519PrivateKey.generate() key_b = Ed25519PrivateKey.generate() pub_b_bytes = key_b.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) pub_b_encoded = encode_pubkey("ed25519", pub_b_bytes) key_b_id = public_key_fingerprint(pub_b_bytes) snap_id = blob_id(b"sv3-snap") # Build a real signature with key_a commit_id = blob_id(b"sv3-commit") ts = now_utc_iso() payload = provenance_payload( commit_id, author="gabriel", agent_id="claude-code", model_id="claude-sonnet-4-6", committed_at=ts, ) sig_a = sign_commit_ed25519(payload, key_a) commit = _make_commit( commit_id=commit_id, snapshot_id=snap_id, committed_at=ts, # Signature from key_a, but public key is key_b → mismatch override_signature=sig_a, override_signer_public_key=pub_b_encoded, override_signer_key_id=key_b_id, ) snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) is_error = result.get("t") == SFRAME_ERROR is_http_error = resp.status_code == 422 assert is_error or is_http_error, ( f"SV3: sig/pubkey mismatch must be rejected. " f"Got status={resp.status_code}, result={result}" ) if is_error: assert result.get("code") == 422, f"expected error code 422, got: {result}" # --------------------------------------------------------------------------- # SV4 — signature present but signer_public_key empty: rejected # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv4_signature_without_public_key_rejected( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV4: A commit with a non-empty signature but empty signer_public_key is rejected. Without the public key we cannot verify the signature — accepting would mean storing unverifiable provenance claims. """ _stub_r2(monkeypatch) repo = await _make_repo(db_session, "SV4 Sig No Pubkey", owner="testuser") private_key = Ed25519PrivateKey.generate() snap_id = blob_id(b"sv4-snap") commit_id = blob_id(b"sv4-commit") ts = now_utc_iso() payload = provenance_payload(commit_id, author="gabriel", committed_at=ts) sig = sign_commit_ed25519(payload, private_key) commit = _make_commit( commit_id=commit_id, snapshot_id=snap_id, committed_at=ts, override_signature=sig, override_signer_public_key="", # no public key → can't verify override_signer_key_id="", ) snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) is_error = result.get("t") == SFRAME_ERROR is_http_error = resp.status_code == 422 assert is_error or is_http_error, ( f"SV4: signature without public key must be rejected. " f"Got status={resp.status_code}, result={result}" ) if is_error: assert result.get("code") == 422, f"expected error code 422, got: {result}" # --------------------------------------------------------------------------- # SV5 — Unsigned commit, require_signed_commits=False: accepted (regression guard) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv5_unsigned_commit_accepted_when_signing_not_required( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV5: Unsigned commits are still accepted when require_signed_commits is off. The signature verification fix must not change existing behavior for repos that do not mandate signed commits. """ _stub_r2(monkeypatch) monkeypatch.setattr( "musehub.services.musehub_wire.settings", _make_settings(require_signed_commits=False), ) repo = await _make_repo(db_session, "SV5 Unsigned OK", owner="testuser") snap_id = blob_id(b"sv5-snap") commit = _make_commit(snapshot_id=snap_id) # no private_key → unsigned snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) assert resp.status_code == 200, f"SV5: unsigned commit should be accepted when not required" assert result.get("ok") is True, f"expected ok=True, got: {result}" # --------------------------------------------------------------------------- # SV6 — Unsigned commit, require_signed_commits=True: rejected (regression guard) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv6_unsigned_commit_rejected_when_signing_required( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV6: Unsigned commits are rejected when require_signed_commits is on.""" _stub_r2(monkeypatch) monkeypatch.setattr( "musehub.services.musehub_wire.settings", _make_settings(require_signed_commits=True), ) repo = await _make_repo(db_session, "SV6 Unsigned Bad", owner="testuser") snap_id = blob_id(b"sv6-snap") commit = _make_commit(snapshot_id=snap_id) # unsigned snap = _make_snapshot(snap_id) body = _header_frame(n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_commits=1) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) is_error = result.get("t") == SFRAME_ERROR is_http_error = resp.status_code == 422 assert is_error or is_http_error, ( f"SV6: unsigned commit must be rejected when required. " f"Got status={resp.status_code}, result={result}" ) # --------------------------------------------------------------------------- # SV7 — Mixed batch: one valid + one forged → entire push rejected # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_sv7_mixed_batch_one_forged_rejects_all( client: AsyncClient, db_session: AsyncSession, auth_headers: StrDict, monkeypatch: pytest.MonkeyPatch, ) -> None: """SV7: A batch with one valid and one forged commit is fully rejected. The push must be atomic — a single bad commit invalidates the entire push, not just the offending commit. """ _stub_r2(monkeypatch) repo = await _make_repo(db_session, "SV7 Mixed Batch", owner="testuser") private_key, raw_pub, pub_str = _new_keypair() key_id = public_key_fingerprint(raw_pub) # Commit 1: legitimately signed snap_id_1 = blob_id(b"sv7-snap-1") commit1 = _make_commit(snapshot_id=snap_id_1, private_key=private_key) # Commit 2: forged signature (garbage bytes but valid format prefix) snap_id_2 = blob_id(b"sv7-snap-2") forged_sig = "ed25519:BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" commit2 = _make_commit( snapshot_id=snap_id_2, override_signature=forged_sig, override_signer_public_key=pub_str, override_signer_key_id=key_id, ) snap1 = _make_snapshot(snap_id_1) snap2 = _make_snapshot(snap_id_2) body = ( _header_frame(n_commits=2) + _commit_pack_frame([commit1, commit2], [snap1, snap2]) + _end_frame(n_commits=2) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**auth_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) result = _decode_result(resp.content) is_error = result.get("t") == SFRAME_ERROR is_http_error = resp.status_code == 422 assert is_error or is_http_error, ( f"SV7: mixed batch with one forged commit must be fully rejected. " f"Got status={resp.status_code}, result={result}" ) if is_error: assert result.get("code") == 422, f"expected error code 422, got: {result}" # --------------------------------------------------------------------------- # Settings factory helper # --------------------------------------------------------------------------- class _SettingsStub: require_signed_commits: bool per_repo_quota_bytes: int trusted_agent_ids: list[str] def _make_settings(*, require_signed_commits: bool = False) -> _SettingsStub: """Return a minimal settings stub that wire_push_stream reads.""" s = _SettingsStub() s.require_signed_commits = require_signed_commits s.per_repo_quota_bytes = 0 # 0 means no quota check s.trusted_agent_ids = [] return s