"""Tests for the MSign per-request authentication layer. MSign is the sole auth mechanism on every protected MuseHub endpoint. Every authenticated request carries: Authorization: MSign handle="{handle}" alg="ed25519" ts={unix_ts} sig="{b64url_sig}" where sig is an Ed25519 signature over the canonical message: "{algorithm}\n{METHOD}\n{host}\n{path_with_query}\n{ts}\n{sha256_hex_of_body}" Coverage: Unit — build_canonical_message determinism and format; _parse_msign_header Integration — require_signed_request / optional_signed_request FastAPI deps E2E — real HTTP stack with real DB identity + key; happy and error paths Security — missing header, wrong scheme, stale timestamp, future timestamp, tampered body, wrong key, unknown handle, revoked key Data — MSignContext fields match the registered identity Stress — 50 sequential signed requests all succeed """ from __future__ import annotations import os import secrets import time from muse.core.types import blob_id, encode_pubkey import msgpack import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from httpx import AsyncClient from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from musehub.core.genesis import compute_identity_id, compute_key_id from musehub.auth.request_signing import ( REPLAY_WINDOW_SECONDS, build_canonical_message, _parse_msign_header, ) from musehub.types.json_types import JSONObject from musehub.crypto.keys import b64url_encode, b64url_decode, key_fingerprint from musehub.db import musehub_models as db from musehub.db.musehub_auth_models import MusehubAuthKey from tests.factories import create_repo as factory_create_repo # ── helpers ──────────────────────────────────────────────────────────────────── def _ed25519_keypair() -> tuple[Ed25519PrivateKey, bytes]: priv = Ed25519PrivateKey.generate() pub = priv.public_key().public_bytes_raw() return priv, pub def _msign_header( priv: Ed25519PrivateKey, handle: str, method: str, path: str, body: bytes, ts: int | None = None, host: str = "test", ) -> str: """Build a valid ``Authorization: MSign …`` header for a test request.""" ts = ts if ts is not None else int(time.time()) canonical = build_canonical_message(method, path, ts, body, host=host) sig_bytes = priv.sign(canonical) sig_b64 = b64url_encode(sig_bytes) return f'MSign handle="{handle}" alg="ed25519" ts={ts} sig="{sig_b64}"' async def _seed_identity( session: AsyncSession, handle: str, priv: Ed25519PrivateKey, pub: bytes, ) -> db.MusehubIdentity: """Insert a MusehubIdentity + MusehubAuthKey row for use in MSign tests.""" identity_id = compute_identity_id(pub) identity = db.MusehubIdentity( identity_id=identity_id, handle=handle, identity_type="human", display_name=handle, ) session.add(identity) await session.flush() public_key_b64 = encode_pubkey("ed25519", pub) key_row = MusehubAuthKey( key_id=compute_key_id(identity_id, public_key_b64), identity_id=identity_id, algorithm="ed25519", public_key_b64=public_key_b64, fingerprint=key_fingerprint(pub), label="test-key", ) session.add(key_row) await session.commit() await session.refresh(identity) return identity def _mp(data: JSONObject) -> bytes: return msgpack.packb(data, use_bin_type=True) def _wire_body() -> bytes: """Minimal valid MWP wire frame: H + C + E with no objects or commits.""" from muse.core.mpack import MuseWireFrameWriter fw = MuseWireFrameWriter() return ( fw.wrap(frame_type="H", payload=_mp({"t": "H", "op": "push", "branch": "main", "n_objects": 0, "n_commits": 0, "have": [], "head": None, "force": False})) + fw.wrap(frame_type="C", payload=_mp({"t": "C", "commits": [], "snapshots": []})) + fw.wrap(frame_type="E", payload=_mp({"t": "E", "n_objects": 0, "n_commits": 0})) ) _WIRE_CT = "application/x-muse-wire" # ── unit: build_canonical_message ───────────────────────────────────────────── class TestBuildCanonicalMessage: def test_deterministic(self) -> None: msg = build_canonical_message("POST", "/foo/bar", 1700000000, b"body", host="test") assert msg == build_canonical_message("POST", "/foo/bar", 1700000000, b"body", host="test") def test_format_is_six_lines(self) -> None: msg = build_canonical_message("GET", "/foo", 1234, b"", host="test").decode() parts = msg.split("\n") assert len(parts) == 6 def test_first_line_is_algorithm(self) -> None: msg = build_canonical_message("DELETE", "/x", 1, b"", host="test").decode() assert msg.startswith("ed25519\n") def test_second_line_is_method(self) -> None: msg = build_canonical_message("DELETE", "/x", 1, b"", host="test").decode() lines = msg.split("\n") assert lines[1] == "DELETE" def test_third_line_is_host(self) -> None: msg = build_canonical_message("GET", "/", 1, b"", host="staging.musehub.ai").decode() lines = msg.split("\n") assert lines[2] == "staging.musehub.ai" def test_fourth_line_is_path(self) -> None: msg = build_canonical_message("POST", "/owner/repo/push?ref=main", 1, b"", host="test").decode() lines = msg.split("\n") assert lines[3] == "/owner/repo/push?ref=main" def test_fifth_line_is_timestamp(self) -> None: ts = 1700000042 msg = build_canonical_message("GET", "/", ts, b"", host="test").decode() assert msg.split("\n")[4] == str(ts) def test_sixth_line_is_sha256_hex_of_body(self) -> None: body = b"hello world" expected = blob_id(body) msg = build_canonical_message("POST", "/", 0, body, host="test").decode() assert msg.split("\n")[5] == expected def test_empty_body_produces_sha256_of_empty(self) -> None: expected = blob_id(b"") msg = build_canonical_message("GET", "/", 0, b"", host="test").decode() assert msg.split("\n")[5] == expected def test_different_method_produces_different_bytes(self) -> None: assert build_canonical_message("GET", "/", 1, b"x") != build_canonical_message("POST", "/", 1, b"x") def test_different_host_produces_different_bytes(self) -> None: assert ( build_canonical_message("GET", "/", 1, b"x", host="localhost") != build_canonical_message("GET", "/", 1, b"x", host="staging.musehub.ai") ) def test_different_path_produces_different_bytes(self) -> None: assert build_canonical_message("GET", "/a", 1, b"x") != build_canonical_message("GET", "/b", 1, b"x") def test_different_ts_produces_different_bytes(self) -> None: assert build_canonical_message("GET", "/", 1, b"x") != build_canonical_message("GET", "/", 2, b"x") def test_different_body_produces_different_bytes(self) -> None: assert build_canonical_message("GET", "/", 1, b"a") != build_canonical_message("GET", "/", 1, b"b") def test_returns_bytes(self) -> None: result = build_canonical_message("GET", "/", 0, b"") assert isinstance(result, bytes) # ── unit: _parse_msign_header ────────────────────────────────────────────────── class TestParseMsignHeader: def test_valid_header_parses(self) -> None: sig = b64url_encode(os.urandom(64)) hdr = f'MSign handle="gabriel" alg="ed25519" ts=1700000000 sig="{sig}"' result = _parse_msign_header(hdr) assert result is not None handle, alg, ts, sig_out = result assert handle == "gabriel" assert alg == "ed25519" assert ts == 1700000000 assert sig_out == sig def test_returns_none_for_bearer(self) -> None: assert _parse_msign_header("Bearer eyJhbGciOiJIUzI1NiJ9.x.y") is None def test_returns_none_for_empty_string(self) -> None: assert _parse_msign_header("") is None def test_returns_none_for_missing_alg(self) -> None: sig = b64url_encode(os.urandom(64)) assert _parse_msign_header(f'MSign handle="gabriel" ts=1700000000 sig="{sig}"') is None def test_returns_none_for_missing_ts(self) -> None: assert _parse_msign_header('MSign handle="gabriel" alg="ed25519" sig="abc"') is None def test_returns_none_for_missing_sig(self) -> None: assert _parse_msign_header('MSign handle="gabriel" alg="ed25519" ts=1234') is None def test_returns_none_for_missing_handle(self) -> None: assert _parse_msign_header('MSign alg="ed25519" ts=1234 sig="abc"') is None def test_handle_with_hyphens_and_underscores(self) -> None: sig = b64url_encode(os.urandom(64)) hdr = f'MSign handle="my-user_123" alg="ed25519" ts=1 sig="{sig}"' result = _parse_msign_header(hdr) assert result is not None assert result[0] == "my-user_123" def test_ts_is_int(self) -> None: sig = b64url_encode(os.urandom(64)) hdr = f'MSign handle="x" alg="ed25519" ts=9999999999 sig="{sig}"' result = _parse_msign_header(hdr) assert result is not None assert isinstance(result[2], int) # ── E2E: require_signed_request (via push/stream endpoint) ──────────────────── @pytest.mark.asyncio async def test_missing_auth_header_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="msign-no-auth", owner="no-auth-user") resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=_wire_body(), headers={"Content-Type": _WIRE_CT}, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_bearer_scheme_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """Bearer tokens are rejected — MSign is the only accepted scheme.""" repo = await factory_create_repo(db_session, slug="msign-bearer-rejected", owner="bearer-user") resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=_wire_body(), headers={ "Content-Type": _WIRE_CT, "Authorization": "Bearer eyJhbGciOiJIUzI1NiJ9.e30.abc123", }, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_malformed_msign_header_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: repo = await factory_create_repo(db_session, slug="msign-malformed", owner="malformed-user") for bad in [ "MSign", "MSign junk", 'MSign handle="x"', 'MSign handle="x" ts=abc sig="def"', ]: resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": bad}, ) assert resp.status_code == 401, f"Expected 401 for {bad!r}, got {resp.status_code}" @pytest.mark.asyncio async def test_stale_timestamp_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """Timestamp older than REPLAY_WINDOW_SECONDS must be rejected.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "stale-ts-user", priv, pub) repo = await factory_create_repo(db_session, slug="msign-stale-ts", owner=identity.handle) stale_ts = int(time.time()) - REPLAY_WINDOW_SECONDS - 5 path = f"/{repo.owner}/{repo.slug}/push/stream" # Streaming: sign with b"" (body hash unknown at signing time) auth = _msign_header(priv, identity.handle, "POST", path, b"", ts=stale_ts) resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code == 401 assert "timestamp" in resp.json().get("detail", "").lower() or "skew" in resp.json().get("detail", "").lower() @pytest.mark.asyncio async def test_future_timestamp_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """Timestamp far in the future must also be rejected (replay prevention).""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "future-ts-user", priv, pub) repo = await factory_create_repo(db_session, slug="msign-future-ts", owner=identity.handle) future_ts = int(time.time()) + REPLAY_WINDOW_SECONDS + 5 path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(priv, identity.handle, "POST", path, b"", ts=future_ts) resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_tampered_signature_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """A modified sig field in the MSign header must be rejected. MWP push is a streaming format — the server always uses b"" for the body hash regardless of actual wire content. Body-level tampering is detected per-frame via content-addressing (each O frame carries a sha256: OID). At the auth layer, the only injectable tamper point is the sig field itself. """ priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "tampered-sig-user", priv, pub) repo = await factory_create_repo(db_session, slug="msign-tampered-sig", owner=identity.handle) path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(priv, identity.handle, "POST", path, b"") # Corrupt the sig field by appending "AAAA" before the closing quote tampered = auth.replace('sig="', 'sig="AAAA', 1) resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": tampered}, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_wrong_key_signature_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """Signature by a different (unregistered) key must be rejected.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "wrong-key-user", priv, pub) repo = await factory_create_repo(db_session, slug="msign-wrong-key", owner=identity.handle) other_priv, _ = _ed25519_keypair() path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(other_priv, identity.handle, "POST", path, b"") resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_unknown_handle_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """A handle that has no identity record must be rejected.""" priv, _ = _ed25519_keypair() repo = await factory_create_repo(db_session, slug="msign-unknown-handle", owner="ghost-user") path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(priv, "ghost-user", "POST", path, b"") resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code == 401 @pytest.mark.asyncio async def test_valid_msign_request_is_accepted( client: AsyncClient, db_session: AsyncSession, ) -> None: """A correctly signed request from a registered identity must not be rejected as 401.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "valid-msign-user", priv, pub) repo = await factory_create_repo( db_session, slug="msign-valid-push", owner=identity.handle ) path = f"/{repo.owner}/{repo.slug}/push/stream" # Streaming: sign with b"" — body hash unknown at signing time auth = _msign_header(priv, identity.handle, "POST", path, b"") resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code != 401, f"Auth rejected a valid MSign request: {resp.text}" @pytest.mark.asyncio async def test_msign_context_contains_correct_identity( client: AsyncClient, db_session: AsyncSession, ) -> None: """The MSignContext injected by require_signed_request must match the seeded identity. Verified indirectly: a non-401 response proves auth passed with the correct handle. If the context had the wrong handle, the push would be rejected as unauthorized (403). """ priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "ctx-identity-user", priv, pub) repo = await factory_create_repo( db_session, slug="msign-ctx-identity", owner=identity.handle ) path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(priv, identity.handle, "POST", path, b"") resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code != 401, f"Auth rejected valid identity: {resp.text}" # ── E2E: optional_signed_request (via refs endpoint) ────────────────────────── @pytest.mark.asyncio async def test_optional_msign_missing_header_still_serves_public_repo( client: AsyncClient, db_session: AsyncSession, ) -> None: """Public repo refs must be served without any Authorization header.""" repo = await factory_create_repo(db_session, slug="msign-optional-public", visibility="public") resp = await client.get(f"/{repo.owner}/{repo.slug}/refs") assert resp.status_code == 200 @pytest.mark.asyncio async def test_optional_msign_valid_header_serves_private_repo( client: AsyncClient, db_session: AsyncSession, ) -> None: """A signed request to a private repo refs endpoint must succeed for the owner.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "optional-msign-owner", priv, pub) repo = await factory_create_repo( db_session, slug="msign-optional-private", owner=identity.handle, visibility="private", ) path = f"/{repo.owner}/{repo.slug}/refs" auth = _msign_header(priv, identity.handle, "GET", path, b"") resp = await client.get(path, headers={"Authorization": auth}) assert resp.status_code == 200 @pytest.mark.asyncio async def test_optional_msign_invalid_header_still_returns_401( client: AsyncClient, db_session: AsyncSession, ) -> None: """Even on optional-auth endpoints, a *present but invalid* MSign header must be rejected. optional_signed_request returns None for *absent* headers, but raises 401 for *present but invalid* headers — you cannot downgrade auth by sending garbage. """ repo = await factory_create_repo( db_session, slug="msign-optional-bad-hdr", visibility="public" ) resp = await client.get( f"/{repo.owner}/{repo.slug}/refs", headers={"Authorization": "MSign garbage-not-valid"}, ) assert resp.status_code == 401 # ── Security: key revocation ────────────────────────────────────────────────── @pytest.mark.asyncio async def test_revoked_key_cannot_authenticate( client: AsyncClient, db_session: AsyncSession, ) -> None: """After a key is revoked, MSign requests signed with that key must be rejected.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "revoke-test-user", priv, pub) repo = await factory_create_repo( db_session, slug="msign-revoke-test", owner=identity.handle ) path = f"/{repo.owner}/{repo.slug}/push/stream" auth = _msign_header(priv, identity.handle, "POST", path, b"") # Before revocation: auth must pass (any non-401) resp_before = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp_before.status_code != 401, resp_before.text # Revoke: delete the MusehubAuthKey row via ORM so the shared session tracks # the deletion properly (bulk DELETE bypasses the identity map). fp = key_fingerprint(pub) key_to_delete = ( await db_session.execute( select(MusehubAuthKey).where(MusehubAuthKey.fingerprint == fp) ) ).scalar_one_or_none() assert key_to_delete is not None, "key not found — setup failed" await db_session.delete(key_to_delete) await db_session.commit() # After revocation: must be rejected (no keys for identity) auth2 = _msign_header(priv, identity.handle, "POST", path, b"") resp_after = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth2}, ) assert resp_after.status_code == 401 @pytest.mark.asyncio async def test_second_key_still_works_after_first_revoked( client: AsyncClient, db_session: AsyncSession, ) -> None: """Multi-key: revoking one key must not affect other registered keys.""" from sqlalchemy import delete as sql_delete priv_a, pub_a = _ed25519_keypair() priv_b, pub_b = _ed25519_keypair() identity = await _seed_identity(db_session, "multi-key-revoke-user", priv_a, pub_a) # Register second key for the same identity pub_key_b64_b = encode_pubkey("ed25519", pub_b) key_b = MusehubAuthKey( key_id=compute_key_id(identity.identity_id, pub_key_b64_b), identity_id=identity.identity_id, algorithm="ed25519", public_key_b64=pub_key_b64_b, fingerprint=key_fingerprint(pub_b), label="key-b", ) db_session.add(key_b) await db_session.commit() repo = await factory_create_repo( db_session, slug="msign-multi-key-revoke", owner=identity.handle ) path = f"/{repo.owner}/{repo.slug}/push/stream" # Revoke key A fp_a = key_fingerprint(pub_a) await db_session.execute( sql_delete(MusehubAuthKey).where(MusehubAuthKey.fingerprint == fp_a) ) await db_session.commit() # Key A rejected auth_a = _msign_header(priv_a, identity.handle, "POST", path, b"") resp_a = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth_a}, ) assert resp_a.status_code == 401 # Key B still works (any non-401) auth_b = _msign_header(priv_b, identity.handle, "POST", path, b"") resp_b = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth_b}, ) assert resp_b.status_code != 401, resp_b.text # ── W8: streaming content-type uses empty body hash ────────────────────────── @pytest.mark.asyncio async def test_wire_content_type_uses_empty_body_hash( client: AsyncClient, db_session: AsyncSession, ) -> None: """application/x-muse-wire push stream must authenticate with sha256("") body hash. Wall 8: we renamed the push Content-Type from application/x-muse-mpack to application/x-muse-wire. The server's MSign auth exempted muse-mpack from body-hash verification (the body is a streaming generator — hash unknown at signing time). After the rename the server fell into the else branch: await request.body() -> sha256(full_body) != sha256("") -> 401. The client always signs streaming requests with sha256("") regardless of content type. The server must recognise application/x-muse-wire as a streaming content type and use b"" for the body hash. """ import struct from muse.core.mpack import MuseWireFrameWriter priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "wire-ct-msign-user", priv, pub) repo = await factory_create_repo( db_session, slug="msign-wire-content-type", owner=identity.handle ) # Build a minimal but valid wire-framed body (H + C + E, no objects). fw = MuseWireFrameWriter() h_payload = _mp({"t": "H", "op": "push", "branch": "main", "n_objects": 0, "n_commits": 0, "have": [], "head": None, "force": False}) c_payload = _mp({"t": "C", "commits": [], "snapshots": []}) e_payload = _mp({"t": "E", "n_objects": 0, "n_commits": 0}) wire_body = ( fw.wrap(frame_type="H", payload=h_payload) + fw.wrap(frame_type="C", payload=c_payload) + fw.wrap(frame_type="E", payload=e_payload) ) path = f"/{repo.owner}/{repo.slug}/push/stream" # Client signs with sha256("") — body is a streaming generator, unknown at signing time. auth = _msign_header(priv, identity.handle, "POST", path, b"") resp = await client.post( path, content=wire_body, headers={ "Content-Type": "application/x-muse-wire", "Authorization": auth, }, ) # Must NOT be 401 — auth must pass. May be 200 (empty push ok) or 422 (bad push state). assert resp.status_code != 401, ( f"Wall 8: server returned 401 for application/x-muse-wire with empty-body-hash signature. " f"Fix: add 'application/x-muse-wire' to the streaming content-type check in " f"musehub/auth/request_signing.py. Detail: {resp.text}" ) # ── Stress: sequential signed requests ──────────────────────────────────────── @pytest.mark.asyncio async def test_25_sequential_signed_requests_all_succeed( client: AsyncClient, db_session: AsyncSession, ) -> None: """25 sequential MSign-authenticated push requests must all be accepted. Capped at 25 to stay within the WIRE_PUSH_LIMIT (30/min) so the test exercises auth correctness without triggering rate limiting. """ priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "stress-msign-user", priv, pub) repo = await factory_create_repo( db_session, slug="msign-stress-test", owner=identity.handle ) path = f"/{repo.owner}/{repo.slug}/push/stream" start = time.perf_counter() for i in range(25): auth = _msign_header(priv, identity.handle, "POST", path, b"") resp = await client.post( path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}, ) assert resp.status_code != 401, f"Request {i} rejected auth: {resp.status_code} {resp.text}" elapsed = time.perf_counter() - start assert elapsed < 10.0, f"25 signed requests took {elapsed:.2f}s — too slow" # ── unit: REPLAY_WINDOW_SECONDS is a positive int ──────────────────────────── def test_replay_window_is_positive_int() -> None: assert isinstance(REPLAY_WINDOW_SECONDS, int) assert REPLAY_WINDOW_SECONDS > 0 def test_replay_window_is_at_least_15_seconds() -> None: """Too small a window would break clients with minor clock drift.""" assert REPLAY_WINDOW_SECONDS >= 15 # ── Performance: canonical message computation latency ──────────────────────── def test_canonical_message_1kb_body_under_1ms() -> None: """build_canonical_message over a 1 KB body must complete in under 1ms. Called on every authenticated request — must be negligible overhead. """ body = os.urandom(1024) samples = 1000 times = [] for _ in range(samples): t0 = time.perf_counter_ns() build_canonical_message("POST", "/owner/repo/push", 1700000000, body) times.append(time.perf_counter_ns() - t0) median_us = sorted(times)[samples // 2] / 1000 assert median_us < 1000, f"Median canonical_message time: {median_us:.1f}µs — exceeds 1ms" def test_canonical_message_1mb_body_under_10ms() -> None: """build_canonical_message over a 1 MB body (SHA-256 of large pack) must be under 10ms.""" body = os.urandom(1024 * 1024) samples = 20 times = [] for _ in range(samples): t0 = time.perf_counter_ns() build_canonical_message("POST", "/owner/repo/push", 1700000000, body) times.append(time.perf_counter_ns() - t0) median_ms = sorted(times)[samples // 2] / 1_000_000 assert median_ms < 10, f"Median canonical_message(1MB) time: {median_ms:.2f}ms — exceeds 10ms" def test_canonical_message_empty_body_is_fast() -> None: """build_canonical_message with an empty body (common for GET requests) is under 100µs. GET requests carry no body — the canonical message is just the SHA-256 of b''. This is the cheapest possible call and must have negligible overhead. """ samples = 1000 times = [] for _ in range(samples): t0 = time.perf_counter_ns() build_canonical_message("GET", "/owner/repo/refs", 1700000000, b"") times.append(time.perf_counter_ns() - t0) median_us = sorted(times)[samples // 2] / 1000 assert median_us < 100, f"Empty-body canonical_message median: {median_us:.1f}µs — exceeds 100µs" # ── Performance: key lookup query efficiency ────────────────────────────────── @pytest.mark.asyncio async def test_verification_with_1_key_under_latency_budget( client: AsyncClient, db_session: AsyncSession, ) -> None: """MSign verification for an identity with 1 key must complete under 200ms.""" priv, pub = _ed25519_keypair() identity = await _seed_identity(db_session, "perf-1key-user", priv, pub) repo = await factory_create_repo(db_session, slug="perf-1key-repo", owner=identity.handle) path = f"/{repo.owner}/{repo.slug}/push/stream" # Warm up (first request includes session setup overhead) auth = _msign_header(priv, identity.handle, "POST", path, b"") await client.post(path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}) t0 = time.perf_counter() auth = _msign_header(priv, identity.handle, "POST", path, b"") resp = await client.post(path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}) elapsed_ms = (time.perf_counter() - t0) * 1000 assert resp.status_code != 401 assert elapsed_ms < 200, f"1-key verification took {elapsed_ms:.0f}ms — exceeds 200ms" @pytest.mark.asyncio async def test_verification_with_5_keys_under_latency_budget( client: AsyncClient, db_session: AsyncSession, ) -> None: """MSign verification for an identity with 5 keys must complete under 200ms. _verify_msign iterates all keys until one verifies. With 5 keys and the correct key last in the list (worst case), latency must still be acceptable. """ priv_correct, pub_correct = _ed25519_keypair() identity = await _seed_identity(db_session, "perf-5key-user", priv_correct, pub_correct) # Add 4 more decoy keys — correct key was inserted first so DB returns it last for i in range(4): _, pub_decoy = _ed25519_keypair() db_session.add(MusehubAuthKey( key_id=secrets.token_hex(16), identity_id=identity.identity_id, algorithm="ed25519", public_key_b64=b64url_encode(pub_decoy), fingerprint=key_fingerprint(pub_decoy), label=f"decoy-{i}", )) await db_session.commit() repo = await factory_create_repo(db_session, slug="perf-5key-repo", owner=identity.handle) path = f"/{repo.owner}/{repo.slug}/push/stream" # Warm up auth = _msign_header(priv_correct, identity.handle, "POST", path, b"") await client.post(path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}) t0 = time.perf_counter() auth = _msign_header(priv_correct, identity.handle, "POST", path, b"") resp = await client.post(path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": auth}) elapsed_ms = (time.perf_counter() - t0) * 1000 assert resp.status_code != 401 assert elapsed_ms < 200, f"5-key verification took {elapsed_ms:.0f}ms — exceeds 200ms" @pytest.mark.asyncio async def test_key_lookup_does_not_degrade_with_more_keys( client: AsyncClient, db_session: AsyncSession, ) -> None: """Verification with 5 keys must not be more than 5× slower than with 1 key. _verify_msign does O(N) crypto iterations across keys, but each Ed25519 verify is fast (~0.1ms). The DB query is a single SELECT — not N queries. The total overhead must remain proportional, not super-linear. """ # Identity A: 1 key priv_a, pub_a = _ed25519_keypair() identity_a = await _seed_identity(db_session, "perf-1key-cmp", priv_a, pub_a) repo_a = await factory_create_repo(db_session, slug="perf-cmp-1key", owner=identity_a.handle) # Identity B: 5 keys (correct key is key B[0], 4 decoys added after) priv_b, pub_b = _ed25519_keypair() identity_b = await _seed_identity(db_session, "perf-5key-cmp", priv_b, pub_b) repo_b = await factory_create_repo(db_session, slug="perf-cmp-5key", owner=identity_b.handle) for i in range(4): _, pub_decoy = _ed25519_keypair() db_session.add(MusehubAuthKey( key_id=secrets.token_hex(16), identity_id=identity_b.identity_id, algorithm="ed25519", public_key_b64=b64url_encode(pub_decoy), fingerprint=key_fingerprint(pub_decoy), label=f"decoy-{i}", )) await db_session.commit() path_a = f"/{repo_a.owner}/{repo_a.slug}/push/stream" path_b = f"/{repo_b.owner}/{repo_b.slug}/push/stream" # Warm up both for priv, identity, path in [(priv_a, identity_a, path_a), (priv_b, identity_b, path_b)]: h = _msign_header(priv, identity.handle, "POST", path, b"") await client.post(path, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": h}) # Measure 1-key identity t0 = time.perf_counter() for _ in range(5): h = _msign_header(priv_a, identity_a.handle, "POST", path_a, b"") r = await client.post(path_a, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": h}) assert r.status_code != 401 time_1key_ms = (time.perf_counter() - t0) * 1000 / 5 # Measure 5-key identity t0 = time.perf_counter() for _ in range(5): h = _msign_header(priv_b, identity_b.handle, "POST", path_b, b"") r = await client.post(path_b, content=_wire_body(), headers={"Content-Type": _WIRE_CT, "Authorization": h}) assert r.status_code != 401 time_5key_ms = (time.perf_counter() - t0) * 1000 / 5 ratio = time_5key_ms / max(time_1key_ms, 1) assert ratio < 5, ( f"5-key verification is {ratio:.1f}× slower than 1-key ({time_5key_ms:.0f}ms vs " f"{time_1key_ms:.0f}ms) — key iteration is super-linear" )