gabriel / musehub public
test_attestations_phase1.py python
1,001 lines 37.9 KB
Raw
sha256:763eb2cb8675073b84c19345b27586d2ed939a9aee97c5479b69f502f1a70eff fix(tests): update test suite to match current implementation Sonnet 4.6 patch 121 days ago
1 """TDD — attestation system Phase 1: claim-type registry, scope, expiry.
2
3 Seven test tiers:
4 T1 Unit — pure functions, no I/O
5 T2 Integration — real DB via db_session fixture
6 T3 End-to-end — FastAPI TestClient round-trips
7 T4 Stress — concurrency / idempotency under load
8 T5 Data integrity — DB invariants, no silent mutation
9 T6 Performance — query latency with index coverage
10 T7 Security — sig replay, cross-protocol, impersonation
11
12 Run a single tier:
13 pytest tests/test_attestations_phase1.py -m tier1 -q --tb=short
14 """
15 from __future__ import annotations
16
17 import asyncio
18 import json
19 import time
20 from datetime import datetime, timedelta, timezone
21 from unittest.mock import AsyncMock, MagicMock
22
23 import pytest
24 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
25 from muse.core.types import encode_pubkey, encode_sig
26 from sqlalchemy.ext.asyncio import AsyncSession
27
28 from musehub.core.genesis import compute_attestation_id
29 from musehub.models.musehub import AttestationRequest, AttestationResponse
30 from tests.factories import create_profile, create_repo
31
32
33 # ---------------------------------------------------------------------------
34 # Shared helpers
35 # ---------------------------------------------------------------------------
36
37 def _utc() -> datetime:
38 return datetime.now(tz=timezone.utc)
39
40
41 def _make_keypair() -> tuple[Ed25519PrivateKey, str]:
42 priv = Ed25519PrivateKey.generate()
43 pub = encode_pubkey("ed25519", priv.public_key().public_bytes_raw())
44 return priv, pub
45
46
47 def _sign(
48 priv: Ed25519PrivateKey,
49 attester: str,
50 subject: str,
51 claim: str,
52 ts: str,
53 scope: str = "identity",
54 scope_ref: str | None = None,
55 ) -> str:
56 parts = ["ATTEST", attester, subject, claim, ts]
57 if scope != "identity" and scope_ref:
58 parts.append(scope_ref)
59 msg = "\n".join(parts).encode()
60 return encode_sig("ed25519", priv.sign(msg))
61
62
63 def _req(
64 attester: str,
65 subject: str,
66 claim: str,
67 priv: Ed25519PrivateKey,
68 pub: str,
69 *,
70 scope: str = "identity",
71 scope_ref: str | None = None,
72 repo_id: str | None = None,
73 commit_id: str | None = None,
74 expires_at: datetime | None = None,
75 issued_at: datetime | None = None,
76 ) -> AttestationRequest:
77 ts = (issued_at or _utc()).isoformat()
78 sig = _sign(priv, attester, subject, claim, ts, scope=scope, scope_ref=scope_ref)
79 return AttestationRequest(
80 attester=attester,
81 subject=subject,
82 claim=claim,
83 signature=sig,
84 attester_public_key=pub,
85 issued_at=issued_at or datetime.fromisoformat(ts),
86 scope=scope,
87 scope_ref=scope_ref,
88 repo_id=repo_id,
89 commit_id=commit_id,
90 expires_at=expires_at,
91 )
92
93
94 # ---------------------------------------------------------------------------
95 # T1 — Unit: claim-type registry
96 # ---------------------------------------------------------------------------
97
98 @pytest.mark.tier1
99 class TestClaimTypeRegistry:
100 """Pure registry lookups — no DB, no I/O."""
101
102 def test_known_type_resolves(self) -> None:
103 from musehub.services.musehub_attestations import get_claim_type
104
105 ct = get_claim_type("human")
106 assert ct is not None
107 assert ct["type_key"] == "human"
108 assert ct["category"] == "identity"
109
110 def test_unknown_type_raises(self) -> None:
111 from musehub.services.musehub_attestations import get_claim_type
112
113 with pytest.raises(ValueError, match="unknown claim type"):
114 get_claim_type("nonsense-claim-xyz")
115
116 def test_deprecated_type_raises_on_issue(self) -> None:
117 """Deprecated types must be rejected at write time."""
118 from musehub.services.musehub_attestations import validate_claim_for_issue
119
120 with pytest.raises(ValueError, match="deprecated"):
121 validate_claim_for_issue("__deprecated_test__")
122
123 def test_all_seed_types_present(self) -> None:
124 from musehub.services.musehub_attestations import list_claim_types
125
126 keys = {ct["type_key"] for ct in list_claim_types()}
127 for expected in (
128 "human", "org", "agent", "spawned-by", "delegate", "trusted",
129 "collab", "co-author", "contractor",
130 "code:reviewed", "code:approved", "deploy:approved",
131 "stems:verified", "mix:approved", "midi:generated", "master:approved",
132 "skill:verified",
133 ):
134 assert expected in keys, f"seed type '{expected}' missing from registry"
135
136 def test_valid_scopes_per_type(self) -> None:
137 from musehub.services.musehub_attestations import get_claim_type
138
139 human = get_claim_type("human")
140 assert human["valid_scopes"] == ["identity"]
141
142 deploy = get_claim_type("deploy:approved")
143 assert "commit" in deploy["valid_scopes"]
144 assert "identity" not in deploy["valid_scopes"]
145
146 stems = get_claim_type("stems:verified")
147 assert "identity" in stems["valid_scopes"]
148 assert "commit" in stems["valid_scopes"]
149
150
151 # ---------------------------------------------------------------------------
152 # T1 — Unit: scope_ref parsing
153 # ---------------------------------------------------------------------------
154
155 @pytest.mark.tier1
156 class TestScopeRefParsing:
157 """Parse and validate the scope_ref string formats."""
158
159 def test_identity_scope_ref_is_handle(self) -> None:
160 from musehub.services.musehub_attestations import parse_scope_ref
161
162 result = parse_scope_ref("identity", "gabriel")
163 assert result["handle"] == "gabriel"
164 assert result["repo_slug"] is None
165 assert result["commit_id"] is None
166
167 def test_repo_scope_ref_parsed(self) -> None:
168 from musehub.services.musehub_attestations import parse_scope_ref
169
170 result = parse_scope_ref("repo", "gabriel/musehub")
171 assert result["handle"] == "gabriel"
172 assert result["repo_slug"] == "musehub"
173 assert result["commit_id"] is None
174
175 def test_commit_scope_ref_parsed(self) -> None:
176 from musehub.services.musehub_attestations import parse_scope_ref
177
178 cid = "sha256:" + "a" * 64
179 result = parse_scope_ref("commit", f"gabriel/musehub@{cid}")
180 assert result["handle"] == "gabriel"
181 assert result["repo_slug"] == "musehub"
182 assert result["commit_id"] == cid
183
184 def test_commit_scope_ref_missing_at_raises(self) -> None:
185 from musehub.services.musehub_attestations import parse_scope_ref
186
187 with pytest.raises(ValueError, match="commit scope_ref must contain '@'"):
188 parse_scope_ref("commit", "gabriel/musehub")
189
190 def test_invalid_scope_raises(self) -> None:
191 from musehub.services.musehub_attestations import parse_scope_ref
192
193 with pytest.raises(ValueError, match="invalid scope"):
194 parse_scope_ref("galaxy", "gabriel")
195
196
197 # ---------------------------------------------------------------------------
198 # T1 — Unit: scope enforcement against claim type valid_scopes
199 # ---------------------------------------------------------------------------
200
201 @pytest.mark.tier1
202 class TestScopeClaimEnforcement:
203
204 def test_human_on_commit_scope_raises(self) -> None:
205 from musehub.services.musehub_attestations import validate_scope_for_claim
206
207 with pytest.raises(ValueError, match="'human' is not valid for scope 'commit'"):
208 validate_scope_for_claim("human", "commit")
209
210 def test_deploy_approved_on_identity_scope_raises(self) -> None:
211 from musehub.services.musehub_attestations import validate_scope_for_claim
212
213 with pytest.raises(ValueError, match="'deploy:approved' is not valid for scope 'identity'"):
214 validate_scope_for_claim("deploy:approved", "identity")
215
216 def test_collab_on_commit_scope_passes(self) -> None:
217 from musehub.services.musehub_attestations import validate_scope_for_claim
218
219 validate_scope_for_claim("collab", "commit") # must not raise
220
221 def test_stems_verified_on_identity_passes(self) -> None:
222 from musehub.services.musehub_attestations import validate_scope_for_claim
223
224 validate_scope_for_claim("stems:verified", "identity") # must not raise
225
226
227 # ---------------------------------------------------------------------------
228 # T1 — Unit: canonical message includes scope_ref
229 # ---------------------------------------------------------------------------
230
231 @pytest.mark.tier1
232 class TestCanonicalMessage:
233
234 def test_identity_scope_message_unchanged(self) -> None:
235 """Legacy identity-scope message must stay compatible with existing sigs."""
236 from musehub.services.musehub_attestations import build_canonical_message
237
238 msg = build_canonical_message(
239 attester="gabriel",
240 subject="aria",
241 claim='{"type":"human"}',
242 issued_at_iso="2026-05-01T00:00:00+00:00",
243 scope="identity",
244 scope_ref=None,
245 )
246 assert msg == b"ATTEST\ngabriel\naria\n{\"type\":\"human\"}\n2026-05-01T00:00:00+00:00"
247
248 def test_commit_scope_appends_scope_ref(self) -> None:
249 from musehub.services.musehub_attestations import build_canonical_message
250
251 cid = "sha256:" + "b" * 64
252 msg = build_canonical_message(
253 attester="gabriel",
254 subject="gabriel/musehub",
255 claim='{"type":"deploy:approved"}',
256 issued_at_iso="2026-05-01T00:00:00+00:00",
257 scope="commit",
258 scope_ref=f"gabriel/musehub@{cid}",
259 )
260 assert f"gabriel/musehub@{cid}".encode() in msg
261
262 def test_deterministic(self) -> None:
263 from musehub.services.musehub_attestations import build_canonical_message
264
265 kwargs = dict(
266 attester="a", subject="b", claim='{"type":"collab"}',
267 issued_at_iso="2026-01-01T00:00:00+00:00",
268 scope="identity", scope_ref=None,
269 )
270 assert build_canonical_message(**kwargs) == build_canonical_message(**kwargs)
271
272
273 # ---------------------------------------------------------------------------
274 # T1 — Unit: AttestationRequest Pydantic validation
275 # ---------------------------------------------------------------------------
276
277 @pytest.mark.tier1
278 class TestAttestationRequestModel:
279
280 def test_scope_default_is_identity(self) -> None:
281 priv, pub = _make_keypair()
282 ts = _utc()
283 sig = _sign(priv, "a", "b", '{"type":"collab"}', ts.isoformat())
284 req = AttestationRequest(
285 attester="a", subject="b", claim='{"type":"collab"}',
286 signature=sig, attester_public_key=pub, issued_at=ts,
287 )
288 assert req.scope == "identity"
289 assert req.scope_ref is None
290 assert req.expires_at is None
291
292 def test_commit_scope_requires_scope_ref(self) -> None:
293 priv, pub = _make_keypair()
294 ts = _utc()
295 sig = _sign(priv, "a", "b", '{"type":"deploy:approved"}', ts.isoformat())
296 with pytest.raises(Exception):
297 AttestationRequest(
298 attester="a", subject="b", claim='{"type":"deploy:approved"}',
299 signature=sig, attester_public_key=pub, issued_at=ts,
300 scope="commit", scope_ref=None, # missing scope_ref for commit
301 )
302
303 def test_expires_at_accepted(self) -> None:
304 priv, pub = _make_keypair()
305 ts = _utc()
306 exp = ts + timedelta(days=90)
307 sig = _sign(priv, "a", "b", '{"type":"collab"}', ts.isoformat())
308 req = AttestationRequest(
309 attester="a", subject="b", claim='{"type":"collab"}',
310 signature=sig, attester_public_key=pub, issued_at=ts,
311 expires_at=exp,
312 )
313 assert req.expires_at == exp
314
315
316 # ---------------------------------------------------------------------------
317 # T2 — Integration: issue with scope and expiry (real DB)
318 # ---------------------------------------------------------------------------
319
320 @pytest.mark.tier2
321 @pytest.mark.asyncio
322 async def test_i1_issue_identity_scope(db_session: AsyncSession) -> None:
323 """Basic identity-scoped attestation round-trip."""
324 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
325
326 priv, pub = _make_keypair()
327 req = _req("gabriel", "aria", '{"type":"human"}', priv, pub)
328 result = await issue_attestation(db_session, req)
329
330 assert result.attester == "gabriel"
331 assert result.subject == "aria"
332 assert result.scope == "identity"
333 assert result.scope_ref is None
334 assert result.expires_at is None
335 assert result.revoked_at is None
336
337 listed = await get_attestations_for_subject(db_session, "aria")
338 assert any(a.attestation_id == result.attestation_id for a in listed.attestations)
339
340
341 @pytest.mark.tier2
342 @pytest.mark.asyncio
343 async def test_i2_issue_commit_scope(db_session: AsyncSession) -> None:
344 """Commit-scoped attestation stores scope_ref and commit_id correctly."""
345 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_commit
346
347 priv, pub = _make_keypair()
348 cid = "sha256:" + "c" * 64
349 scope_ref = f"gabriel/musehub@{cid}"
350
351 req = _req(
352 "gabriel", "gabriel/musehub", '{"type":"deploy:approved"}', priv, pub,
353 scope="commit", scope_ref=scope_ref, commit_id=cid,
354 )
355 result = await issue_attestation(db_session, req)
356
357 assert result.scope == "commit"
358 assert result.scope_ref == scope_ref
359 assert result.commit_id == cid
360
361 by_commit = await get_attestations_for_commit(db_session, cid)
362 assert any(a.attestation_id == result.attestation_id for a in by_commit.attestations)
363
364
365 @pytest.mark.tier2
366 @pytest.mark.asyncio
367 async def test_i3_commit_scope_excluded_from_identity_query(db_session: AsyncSession) -> None:
368 """Commit-scoped attestations must not appear in identity-scoped profile queries."""
369 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
370
371 priv, pub = _make_keypair()
372 cid = "sha256:" + "d" * 64
373 req = _req(
374 "gabriel", "gabriel/musehub", '{"type":"deploy:approved"}', priv, pub,
375 scope="commit", scope_ref=f"gabriel/musehub@{cid}", commit_id=cid,
376 )
377 await issue_attestation(db_session, req)
378
379 # subject for commit scope is repo slug, not a handle — should not appear
380 listed = await get_attestations_for_subject(db_session, "gabriel/musehub")
381 assert all(a.scope == "identity" for a in listed.attestations)
382
383
384 @pytest.mark.tier2
385 @pytest.mark.asyncio
386 async def test_i4_expired_attestation_excluded_by_default(db_session: AsyncSession) -> None:
387 """Attestations past expires_at are excluded from default queries."""
388 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
389
390 priv, pub = _make_keypair()
391 past = _utc() - timedelta(seconds=1)
392 req = _req("gabriel", "aria", '{"type":"contractor"}', priv, pub, expires_at=past)
393 result = await issue_attestation(db_session, req)
394
395 listed = await get_attestations_for_subject(db_session, "aria")
396 assert all(a.attestation_id != result.attestation_id for a in listed.attestations)
397
398
399 @pytest.mark.tier2
400 @pytest.mark.asyncio
401 async def test_i5_expired_attestation_included_with_flag(db_session: AsyncSession) -> None:
402 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
403
404 priv, pub = _make_keypair()
405 past = _utc() - timedelta(seconds=1)
406 req = _req("gabriel", "aria", '{"type":"contractor"}', priv, pub, expires_at=past)
407 result = await issue_attestation(db_session, req)
408
409 listed = await get_attestations_for_subject(db_session, "aria", include_expired=True)
410 assert any(a.attestation_id == result.attestation_id for a in listed.attestations)
411
412
413 @pytest.mark.tier2
414 @pytest.mark.asyncio
415 async def test_i6_unknown_claim_type_rejected(db_session: AsyncSession) -> None:
416 """Registry enforcement: unknown claim type raises ValueError before DB write."""
417 from musehub.services.musehub_attestations import issue_attestation
418
419 priv, pub = _make_keypair()
420 req = _req("gabriel", "aria", '{"type":"galaxy-brain"}', priv, pub)
421
422 with pytest.raises(ValueError, match="unknown claim type"):
423 await issue_attestation(db_session, req)
424
425
426 @pytest.mark.tier2
427 @pytest.mark.asyncio
428 async def test_i7_wrong_scope_for_claim_type_rejected(db_session: AsyncSession) -> None:
429 """`human` claim type only valid for identity scope — commit scope raises."""
430 from musehub.services.musehub_attestations import issue_attestation
431
432 priv, pub = _make_keypair()
433 cid = "sha256:" + "e" * 64
434 req = _req(
435 "gabriel", "gabriel/musehub", '{"type":"human"}', priv, pub,
436 scope="commit", scope_ref=f"gabriel/musehub@{cid}", commit_id=cid,
437 )
438 with pytest.raises(ValueError, match="not valid for scope"):
439 await issue_attestation(db_session, req)
440
441
442 @pytest.mark.tier2
443 @pytest.mark.asyncio
444 async def test_i8_idempotent_issue(db_session: AsyncSession) -> None:
445 """Same canonical payload issued twice returns existing row, inserts only once."""
446 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
447
448 priv, pub = _make_keypair()
449 ts = _utc()
450 req = _req("gabriel", "aria", '{"type":"trusted"}', priv, pub, issued_at=ts)
451
452 r1 = await issue_attestation(db_session, req)
453 r2 = await issue_attestation(db_session, req)
454
455 assert r1.attestation_id == r2.attestation_id
456 listed = await get_attestations_for_subject(db_session, "aria")
457 matching = [a for a in listed.attestations if a.attestation_id == r1.attestation_id]
458 assert len(matching) == 1
459
460
461 @pytest.mark.tier2
462 @pytest.mark.asyncio
463 async def test_i9_repo_scope_stored_and_retrieved(db_session: AsyncSession) -> None:
464 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_repo
465
466 priv, pub = _make_keypair()
467 req = _req(
468 "gabriel", "gabriel/musehub", '{"type":"code:reviewed"}', priv, pub,
469 scope="repo", scope_ref="gabriel/musehub",
470 )
471 result = await issue_attestation(db_session, req)
472 assert result.scope == "repo"
473
474 by_repo = await get_attestations_for_repo(db_session, "gabriel/musehub")
475 assert any(a.attestation_id == result.attestation_id for a in by_repo.attestations)
476
477
478 # ---------------------------------------------------------------------------
479 # T2 — Integration: DB-backed claim type registry
480 # ---------------------------------------------------------------------------
481
482 @pytest.mark.tier2
483 @pytest.mark.asyncio
484 async def test_i10_registry_seed_types_in_db(db_session: AsyncSession) -> None:
485 """Seed claim types are present in the DB registry after migration."""
486 from musehub.services.musehub_attestations import get_claim_type_from_db
487
488 ct = await get_claim_type_from_db(db_session, "human")
489 assert ct is not None
490 assert ct["category"] == "identity"
491
492
493 @pytest.mark.tier2
494 @pytest.mark.asyncio
495 async def test_i11_add_new_claim_type(db_session: AsyncSession) -> None:
496 """New claim types can be added to the DB registry at runtime."""
497 from musehub.services.musehub_attestations import add_claim_type, get_claim_type_from_db
498
499 await add_claim_type(
500 db_session,
501 type_key="test:custom",
502 category="collab",
503 label="Custom Test",
504 description="A test-only claim type.",
505 valid_scopes=["identity", "commit"],
506 )
507
508 ct = await get_claim_type_from_db(db_session, "test:custom")
509 assert ct is not None
510 assert ct["label"] == "Custom Test"
511
512
513 @pytest.mark.tier2
514 @pytest.mark.asyncio
515 async def test_i12_deprecated_type_rejected(db_session: AsyncSession) -> None:
516 """After deprecating a type, issue_attestation raises ValueError."""
517 from musehub.services.musehub_attestations import (
518 add_claim_type, deprecate_claim_type, issue_attestation,
519 )
520
521 await add_claim_type(
522 db_session,
523 type_key="test:to-deprecate",
524 category="collab",
525 label="Deprecated Soon",
526 description="Will be deprecated.",
527 valid_scopes=["identity"],
528 )
529 await deprecate_claim_type(db_session, "test:to-deprecate")
530
531 priv, pub = _make_keypair()
532 req = _req("gabriel", "aria", '{"type":"test:to-deprecate"}', priv, pub)
533 with pytest.raises(ValueError, match="deprecated"):
534 await issue_attestation(db_session, req)
535
536
537 # ---------------------------------------------------------------------------
538 # T3 — End-to-end: API round-trips via TestClient
539 # ---------------------------------------------------------------------------
540
541 @pytest.fixture
542 def real_keypair():
543 """A fresh Ed25519 keypair for E2E signing."""
544 return _make_keypair()
545
546
547 @pytest.mark.tier3
548 class TestAttestationAPIRoundTrip:
549
550 @pytest.mark.asyncio
551 async def test_e2e_create_and_list_identity_scope(self, client, real_keypair) -> None:
552 """POST /api/profiles/{handle}/attestations → GET lists it."""
553 priv, pub = real_keypair
554 ts = _utc()
555 claim = '{"type":"collab"}'
556 sig = _sign(priv, "gabriel", "aria", claim, ts.isoformat())
557
558 resp = await client.post(
559 "/api/profiles/aria/attestations",
560 json={
561 "attester": "gabriel",
562 "subject": "aria",
563 "claim": claim,
564 "signature": sig,
565 "attester_public_key": pub,
566 "issued_at": ts.isoformat(),
567 "scope": "identity",
568 },
569 )
570 assert resp.status_code == 201
571 aid = resp.json()["attestationId"]
572
573 resp = await client.get("/api/profiles/aria/attestations")
574 assert resp.status_code == 200
575 ids = [a["attestationId"] for a in resp.json()["attestations"]]
576 assert aid in ids
577
578 @pytest.mark.asyncio
579 async def test_e2e_commit_scope_not_in_profile_list(self, client, real_keypair) -> None:
580 """Commit-scoped attestations don't appear in identity profile listing."""
581 priv, pub = real_keypair
582 ts = _utc()
583 cid = "sha256:" + "f" * 64
584 claim = '{"type":"deploy:approved"}'
585 sig = _sign(priv, "gabriel", "gabriel/musehub", claim, ts.isoformat(),
586 scope="commit", scope_ref=f"gabriel/musehub@{cid}")
587
588 resp = await client.post(
589 "/api/profiles/gabriel/attestations",
590 json={
591 "attester": "gabriel",
592 "subject": "gabriel/musehub",
593 "claim": claim,
594 "signature": sig,
595 "attester_public_key": pub,
596 "issued_at": ts.isoformat(),
597 "scope": "commit",
598 "scope_ref": f"gabriel/musehub@{cid}",
599 "commit_id": cid,
600 },
601 )
602 assert resp.status_code == 201
603
604 resp = await client.get("/api/profiles/gabriel/attestations")
605 for a in resp.json()["attestations"]:
606 assert a.get("scope", "identity") == "identity"
607
608 @pytest.mark.asyncio
609 async def test_e2e_unknown_type_returns_400(self, client, real_keypair) -> None:
610 priv, pub = real_keypair
611 ts = _utc()
612 claim = '{"type":"made-up-claim"}'
613 sig = _sign(priv, "gabriel", "aria", claim, ts.isoformat())
614
615 resp = await client.post(
616 "/api/profiles/aria/attestations",
617 json={
618 "attester": "gabriel", "subject": "aria", "claim": claim,
619 "signature": sig, "attester_public_key": pub,
620 "issued_at": ts.isoformat(), "scope": "identity",
621 },
622 )
623 assert resp.status_code == 400
624
625 @pytest.mark.asyncio
626 async def test_e2e_claim_types_endpoint(self, client) -> None:
627 """GET /api/profiles/attestation-types returns the seeded registry."""
628 resp = await client.get("/api/profiles/attestation-types")
629 assert resp.status_code == 200
630 keys = {ct["typeKey"] for ct in resp.json()["claimTypes"]}
631 assert "human" in keys
632 assert "deploy:approved" in keys
633
634 @pytest.mark.asyncio
635 async def test_e2e_revoke_removes_from_listing(self, client, real_keypair) -> None:
636 priv, pub = real_keypair
637 ts = _utc()
638 claim = '{"type":"trusted"}'
639 sig = _sign(priv, "gabriel", "aria", claim, ts.isoformat())
640
641 create = await client.post(
642 "/api/profiles/aria/attestations",
643 json={
644 "attester": "gabriel", "subject": "aria", "claim": claim,
645 "signature": sig, "attester_public_key": pub,
646 "issued_at": ts.isoformat(), "scope": "identity",
647 },
648 )
649 assert create.status_code == 201
650 aid = create.json()["attestationId"]
651
652 revoke = await client.delete(f"/api/profiles/aria/attestations/{aid}",
653 params={"revoker": "gabriel"})
654 assert revoke.status_code == 200
655
656 listed = await client.get("/api/profiles/aria/attestations")
657 ids = [a["attestationId"] for a in listed.json()["attestations"]]
658 assert aid not in ids
659
660
661 # ---------------------------------------------------------------------------
662 # T4 — Stress: idempotency under concurrent writes
663 # ---------------------------------------------------------------------------
664
665 @pytest.mark.tier4
666 @pytest.mark.asyncio
667 async def test_s1_concurrent_identical_issue_is_idempotent(session_factory) -> None:
668 """50 concurrent issues of the same payload → exactly 1 row."""
669 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
670
671 priv, pub = _make_keypair()
672 ts = _utc()
673 req = _req("gabriel", "aria-stress", '{"type":"collab"}', priv, pub, issued_at=ts)
674
675 async def _do() -> AttestationResponse:
676 async with session_factory() as sess:
677 return await issue_attestation(sess, req)
678
679 results = await asyncio.gather(*[_do() for _ in range(50)])
680 ids = {r.attestation_id for r in results}
681 assert len(ids) == 1
682
683 async with session_factory() as check_sess:
684 listed = await get_attestations_for_subject(check_sess, "aria-stress")
685 matching = [a for a in listed.attestations if a.attester == "gabriel"]
686 assert len(matching) == 1
687
688
689 @pytest.mark.tier4
690 @pytest.mark.asyncio
691 async def test_s2_many_distinct_attestations(session_factory) -> None:
692 """100 distinct attestations insert without error (max 10 concurrent connections)."""
693 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
694
695 sem = asyncio.Semaphore(10)
696
697 async def _issue(i: int) -> None:
698 priv, pub = _make_keypair()
699 req = _req(f"attester-{i:04d}", "aria-many", '{"type":"trusted"}', priv, pub)
700 async with sem:
701 async with session_factory() as sess:
702 await issue_attestation(sess, req)
703
704 await asyncio.gather(*[_issue(i) for i in range(100)])
705 async with session_factory() as check_sess:
706 listed = await get_attestations_for_subject(check_sess, "aria-many")
707 assert listed.total >= 100
708
709
710 # ---------------------------------------------------------------------------
711 # T5 — Data integrity
712 # ---------------------------------------------------------------------------
713
714 @pytest.mark.tier5
715 @pytest.mark.asyncio
716 async def test_d1_attestation_id_unique_constraint(db_session: AsyncSession) -> None:
717 """Duplicate attestation_id raises IntegrityError, not silent overwrite."""
718 import sqlalchemy.exc
719 from musehub.db.musehub_models import MusehubAttestation
720
721 priv, pub = _make_keypair()
722 ts = _utc()
723 claim = '{"type":"collab"}'
724 aid = compute_attestation_id("gabriel", "aria", claim, ts.isoformat())
725
726 row = MusehubAttestation(
727 attestation_id=aid,
728 attester="gabriel",
729 subject="aria",
730 claim=claim,
731 signature=_sign(priv, "gabriel", "aria", claim, ts.isoformat()),
732 attester_public_key=pub,
733 issued_at=ts,
734 scope="identity",
735 )
736 db_session.add(row)
737 await db_session.commit()
738 db_session.expunge(row)
739
740 dup = MusehubAttestation(
741 attestation_id=aid, # same PK
742 attester="gabriel",
743 subject="aria",
744 claim=claim,
745 signature=_sign(priv, "gabriel", "aria", claim, ts.isoformat()),
746 attester_public_key=pub,
747 issued_at=ts,
748 scope="identity",
749 )
750 db_session.add(dup)
751 with pytest.raises(sqlalchemy.exc.IntegrityError):
752 await db_session.commit()
753
754
755 @pytest.mark.tier5
756 @pytest.mark.asyncio
757 async def test_d2_revoked_at_set_not_deleted(db_session: AsyncSession) -> None:
758 """Revocation sets revoked_at; the row is never deleted."""
759 from musehub.services.musehub_attestations import issue_attestation, revoke_attestation
760 from sqlalchemy import select
761 from musehub.db.musehub_models import MusehubAttestation
762
763 priv, pub = _make_keypair()
764 req = _req("gabriel", "aria", '{"type":"trusted"}', priv, pub)
765 result = await issue_attestation(db_session, req)
766
767 await revoke_attestation(db_session, result.attestation_id, revoker="gabriel")
768
769 row = (await db_session.execute(
770 select(MusehubAttestation).where(
771 MusehubAttestation.attestation_id == result.attestation_id
772 )
773 )).scalar_one()
774 assert row.revoked_at is not None
775 assert row.attester == "gabriel" # row intact
776
777
778 @pytest.mark.tier5
779 @pytest.mark.asyncio
780 async def test_d3_scope_columns_non_nullable_defaults(db_session: AsyncSession) -> None:
781 """scope column has non-null default of 'identity'; scope_ref and commit_id are nullable."""
782 from musehub.services.musehub_attestations import issue_attestation
783 from sqlalchemy import select
784 from musehub.db.musehub_models import MusehubAttestation
785
786 priv, pub = _make_keypair()
787 req = _req("gabriel", "aria", '{"type":"collab"}', priv, pub)
788 result = await issue_attestation(db_session, req)
789
790 row = (await db_session.execute(
791 select(MusehubAttestation).where(
792 MusehubAttestation.attestation_id == result.attestation_id
793 )
794 )).scalar_one()
795 assert row.scope == "identity"
796 assert row.scope_ref is None
797 assert row.commit_id is None
798 assert row.expires_at is None
799
800
801 @pytest.mark.tier5
802 @pytest.mark.asyncio
803 async def test_d4_claim_stored_verbatim(db_session: AsyncSession) -> None:
804 """claim JSON is stored byte-for-byte; no normalisation changes claim_type derivation."""
805 from musehub.services.musehub_attestations import issue_attestation
806 from sqlalchemy import select
807 from musehub.db.musehub_models import MusehubAttestation
808
809 priv, pub = _make_keypair()
810 raw_claim = '{"type":"collab","note":"verbatim"}'
811 req = _req("gabriel", "aria", raw_claim, priv, pub)
812 result = await issue_attestation(db_session, req)
813
814 row = (await db_session.execute(
815 select(MusehubAttestation).where(
816 MusehubAttestation.attestation_id == result.attestation_id
817 )
818 )).scalar_one()
819 assert row.claim == raw_claim
820
821
822 # ---------------------------------------------------------------------------
823 # T6 — Performance
824 # ---------------------------------------------------------------------------
825
826 @pytest.mark.tier6
827 @pytest.mark.asyncio
828 async def test_p1_subject_query_under_50ms_with_500_rows(db_session: AsyncSession) -> None:
829 """get_attestations_for_subject with 500 rows completes in < 50ms."""
830 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
831
832 async def _bulk() -> None:
833 for i in range(500):
834 priv, pub = _make_keypair()
835 req = _req(f"bulk-{i:04d}", "perf-subject", '{"type":"trusted"}', priv, pub)
836 await issue_attestation(db_session, req)
837
838 await _bulk()
839
840 t0 = time.monotonic()
841 result = await get_attestations_for_subject(db_session, "perf-subject")
842 elapsed_ms = (time.monotonic() - t0) * 1000
843
844 assert result.total >= 500
845 assert elapsed_ms < 50, f"query took {elapsed_ms:.1f}ms — expected < 50ms"
846
847
848 @pytest.mark.tier6
849 @pytest.mark.asyncio
850 async def test_p2_commit_query_under_20ms(db_session: AsyncSession) -> None:
851 """get_attestations_for_commit with index returns in < 20ms."""
852 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_commit
853
854 cid = "sha256:" + "9" * 64
855 for i in range(50):
856 priv, pub = _make_keypair()
857 req = _req(
858 f"perf-att-{i:03d}", "gabriel/musehub",
859 '{"type":"deploy:approved"}', priv, pub,
860 scope="commit", scope_ref=f"gabriel/musehub@{cid}", commit_id=cid,
861 )
862 await issue_attestation(db_session, req)
863
864 t0 = time.monotonic()
865 result = await get_attestations_for_commit(db_session, cid)
866 elapsed_ms = (time.monotonic() - t0) * 1000
867
868 assert result.total >= 50
869 assert elapsed_ms < 50, f"commit query took {elapsed_ms:.1f}ms — expected < 50ms"
870
871
872 # ---------------------------------------------------------------------------
873 # T7 — Security
874 # ---------------------------------------------------------------------------
875
876 @pytest.mark.tier7
877 @pytest.mark.asyncio
878 async def test_sec1_cross_protocol_replay_rejected(db_session: AsyncSession) -> None:
879 """MSign-prefixed message cannot satisfy attestation signature check."""
880 from musehub.services.musehub_attestations import issue_attestation
881
882 priv, pub = _make_keypair()
883 ts = _utc()
884 claim = '{"type":"collab"}'
885 # Sign with MSign prefix instead of ATTEST prefix
886 msign_msg = f"MUSE-SIGN-V1\ngabriel\naria\n{claim}\n{ts.isoformat()}".encode()
887 bad_sig = encode_sig("ed25519", priv.sign(msign_msg))
888
889 req = AttestationRequest(
890 attester="gabriel", subject="aria", claim=claim,
891 signature=bad_sig, attester_public_key=pub, issued_at=ts,
892 )
893 with pytest.raises(ValueError, match="Invalid attestation signature"):
894 await issue_attestation(db_session, req)
895
896
897 @pytest.mark.tier7
898 @pytest.mark.asyncio
899 async def test_sec2_attester_impersonation_rejected(db_session: AsyncSession) -> None:
900 """Claiming attester=gabriel but signing with a different key → rejected."""
901 from musehub.services.musehub_attestations import issue_attestation
902
903 _, real_pub = _make_keypair() # gabriel's real key (pub only)
904 evil_priv, _ = _make_keypair() # attacker's key
905
906 ts = _utc()
907 claim = '{"type":"human"}'
908 # Attacker signs with own key but presents gabriel's pubkey
909 evil_sig = _sign(evil_priv, "gabriel", "aria", claim, ts.isoformat())
910
911 req = AttestationRequest(
912 attester="gabriel", subject="aria", claim=claim,
913 signature=evil_sig, attester_public_key=real_pub, issued_at=ts,
914 )
915 with pytest.raises(ValueError, match="Invalid attestation signature"):
916 await issue_attestation(db_session, req)
917
918
919 @pytest.mark.tier7
920 @pytest.mark.asyncio
921 async def test_sec3_tampered_claim_rejected(db_session: AsyncSession) -> None:
922 """Signature over original claim does not verify for a mutated claim."""
923 from musehub.services.musehub_attestations import issue_attestation
924
925 priv, pub = _make_keypair()
926 ts = _utc()
927 original_claim = '{"type":"collab"}'
928 tampered_claim = '{"type":"human"}' # different type
929 sig = _sign(priv, "gabriel", "aria", original_claim, ts.isoformat())
930
931 req = AttestationRequest(
932 attester="gabriel", subject="aria", claim=tampered_claim,
933 signature=sig, attester_public_key=pub, issued_at=ts,
934 )
935 with pytest.raises(ValueError, match="Invalid attestation signature"):
936 await issue_attestation(db_session, req)
937
938
939 @pytest.mark.tier7
940 @pytest.mark.asyncio
941 async def test_sec4_revoker_impersonation_rejected(db_session: AsyncSession) -> None:
942 """Wrong revoker cannot retract someone else's attestation."""
943 from musehub.services.musehub_attestations import issue_attestation, revoke_attestation
944
945 priv, pub = _make_keypair()
946 req = _req("gabriel", "aria", '{"type":"trusted"}', priv, pub)
947 result = await issue_attestation(db_session, req)
948
949 with pytest.raises(ValueError, match="only the attester can revoke"):
950 await revoke_attestation(db_session, result.attestation_id, revoker="evil-actor")
951
952
953 @pytest.mark.tier7
954 @pytest.mark.asyncio
955 async def test_sec5_expired_attestation_excluded_without_flag(db_session: AsyncSession) -> None:
956 """Expired attestation is excluded from live queries even if not explicitly revoked."""
957 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
958
959 priv, pub = _make_keypair()
960 past = _utc() - timedelta(hours=1)
961 req = _req("gabriel", "aria-exp", '{"type":"contractor"}', priv, pub, expires_at=past)
962 result = await issue_attestation(db_session, req)
963
964 listed = await get_attestations_for_subject(db_session, "aria-exp")
965 assert all(a.attestation_id != result.attestation_id for a in listed.attestations)
966 assert result.revoked_at is None # not revoked — just expired
967
968
969 @pytest.mark.tier7
970 @pytest.mark.asyncio
971 async def test_sec6_key_rotation_old_attestations_still_verifiable(db_session: AsyncSession) -> None:
972 """attester_public_key is stored per-attestation; key rotation doesn't break old records."""
973 from musehub.services.musehub_attestations import issue_attestation, verify_stored_attestation
974
975 old_priv, old_pub = _make_keypair()
976 req = _req("gabriel", "aria", '{"type":"collab"}', old_priv, old_pub)
977 result = await issue_attestation(db_session, req)
978
979 # Simulate key rotation — new key exists but old attestation used old key
980 _new_priv, _new_pub = _make_keypair()
981
982 # Verify using the stored public key (not the current key)
983 ok, reason = await verify_stored_attestation(db_session, result.attestation_id)
984 assert ok, f"old attestation failed verification after key rotation: {reason}"
985
986
987 @pytest.mark.tier7
988 @pytest.mark.asyncio
989 async def test_sec7_sql_metachar_in_claim_stored_safely(db_session: AsyncSession) -> None:
990 """SQL metacharacters in claim payload are stored verbatim without injection."""
991 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
992
993 priv, pub = _make_keypair()
994 evil_claim = '{"type":"collab","note":"Robert\'); DROP TABLE musehub_attestations;--"}'
995 req = _req("gabriel", "aria-sql", evil_claim, priv, pub)
996 result = await issue_attestation(db_session, req)
997
998 listed = await get_attestations_for_subject(db_session, "aria-sql")
999 found = next((a for a in listed.attestations if a.attestation_id == result.attestation_id), None)
1000 assert found is not None
1001 assert "DROP TABLE" in found.claim # stored verbatim, not executed
File History 1 commit
sha256:763eb2cb8675073b84c19345b27586d2ed939a9aee97c5479b69f502f1a70eff fix(tests): update test suite to match current implementation Sonnet 4.6 patch 121 days ago