gabriel / musehub public
test_attestations_phase1.py python
1,000 lines 37.9 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 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
739 dup = MusehubAttestation(
740 attestation_id=aid, # same PK
741 attester="gabriel",
742 subject="aria",
743 claim=claim,
744 signature=_sign(priv, "gabriel", "aria", claim, ts.isoformat()),
745 attester_public_key=pub,
746 issued_at=ts,
747 scope="identity",
748 )
749 db_session.add(dup)
750 with pytest.raises(sqlalchemy.exc.IntegrityError):
751 await db_session.commit()
752
753
754 @pytest.mark.tier5
755 @pytest.mark.asyncio
756 async def test_d2_revoked_at_set_not_deleted(db_session: AsyncSession) -> None:
757 """Revocation sets revoked_at; the row is never deleted."""
758 from musehub.services.musehub_attestations import issue_attestation, revoke_attestation
759 from sqlalchemy import select
760 from musehub.db.musehub_models import MusehubAttestation
761
762 priv, pub = _make_keypair()
763 req = _req("gabriel", "aria", '{"type":"trusted"}', priv, pub)
764 result = await issue_attestation(db_session, req)
765
766 await revoke_attestation(db_session, result.attestation_id, revoker="gabriel")
767
768 row = (await db_session.execute(
769 select(MusehubAttestation).where(
770 MusehubAttestation.attestation_id == result.attestation_id
771 )
772 )).scalar_one()
773 assert row.revoked_at is not None
774 assert row.attester == "gabriel" # row intact
775
776
777 @pytest.mark.tier5
778 @pytest.mark.asyncio
779 async def test_d3_scope_columns_non_nullable_defaults(db_session: AsyncSession) -> None:
780 """scope column has non-null default of 'identity'; scope_ref and commit_id are nullable."""
781 from musehub.services.musehub_attestations import issue_attestation
782 from sqlalchemy import select
783 from musehub.db.musehub_models import MusehubAttestation
784
785 priv, pub = _make_keypair()
786 req = _req("gabriel", "aria", '{"type":"collab"}', priv, pub)
787 result = await issue_attestation(db_session, req)
788
789 row = (await db_session.execute(
790 select(MusehubAttestation).where(
791 MusehubAttestation.attestation_id == result.attestation_id
792 )
793 )).scalar_one()
794 assert row.scope == "identity"
795 assert row.scope_ref is None
796 assert row.commit_id is None
797 assert row.expires_at is None
798
799
800 @pytest.mark.tier5
801 @pytest.mark.asyncio
802 async def test_d4_claim_stored_verbatim(db_session: AsyncSession) -> None:
803 """claim JSON is stored byte-for-byte; no normalisation changes claim_type derivation."""
804 from musehub.services.musehub_attestations import issue_attestation
805 from sqlalchemy import select
806 from musehub.db.musehub_models import MusehubAttestation
807
808 priv, pub = _make_keypair()
809 raw_claim = '{"type":"collab","note":"verbatim"}'
810 req = _req("gabriel", "aria", raw_claim, priv, pub)
811 result = await issue_attestation(db_session, req)
812
813 row = (await db_session.execute(
814 select(MusehubAttestation).where(
815 MusehubAttestation.attestation_id == result.attestation_id
816 )
817 )).scalar_one()
818 assert row.claim == raw_claim
819
820
821 # ---------------------------------------------------------------------------
822 # T6 — Performance
823 # ---------------------------------------------------------------------------
824
825 @pytest.mark.tier6
826 @pytest.mark.asyncio
827 async def test_p1_subject_query_under_50ms_with_500_rows(db_session: AsyncSession) -> None:
828 """get_attestations_for_subject with 500 rows completes in < 50ms."""
829 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
830
831 async def _bulk() -> None:
832 for i in range(500):
833 priv, pub = _make_keypair()
834 req = _req(f"bulk-{i:04d}", "perf-subject", '{"type":"trusted"}', priv, pub)
835 await issue_attestation(db_session, req)
836
837 await _bulk()
838
839 t0 = time.monotonic()
840 result = await get_attestations_for_subject(db_session, "perf-subject")
841 elapsed_ms = (time.monotonic() - t0) * 1000
842
843 assert result.total >= 500
844 assert elapsed_ms < 50, f"query took {elapsed_ms:.1f}ms — expected < 50ms"
845
846
847 @pytest.mark.tier6
848 @pytest.mark.asyncio
849 async def test_p2_commit_query_under_20ms(db_session: AsyncSession) -> None:
850 """get_attestations_for_commit with index returns in < 20ms."""
851 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_commit
852
853 cid = "sha256:" + "9" * 64
854 for i in range(50):
855 priv, pub = _make_keypair()
856 req = _req(
857 f"perf-att-{i:03d}", "gabriel/musehub",
858 '{"type":"deploy:approved"}', priv, pub,
859 scope="commit", scope_ref=f"gabriel/musehub@{cid}", commit_id=cid,
860 )
861 await issue_attestation(db_session, req)
862
863 t0 = time.monotonic()
864 result = await get_attestations_for_commit(db_session, cid)
865 elapsed_ms = (time.monotonic() - t0) * 1000
866
867 assert result.total >= 50
868 assert elapsed_ms < 50, f"commit query took {elapsed_ms:.1f}ms — expected < 50ms"
869
870
871 # ---------------------------------------------------------------------------
872 # T7 — Security
873 # ---------------------------------------------------------------------------
874
875 @pytest.mark.tier7
876 @pytest.mark.asyncio
877 async def test_sec1_cross_protocol_replay_rejected(db_session: AsyncSession) -> None:
878 """MSign-prefixed message cannot satisfy attestation signature check."""
879 from musehub.services.musehub_attestations import issue_attestation
880
881 priv, pub = _make_keypair()
882 ts = _utc()
883 claim = '{"type":"collab"}'
884 # Sign with MSign prefix instead of ATTEST prefix
885 msign_msg = f"MUSE-SIGN-V1\ngabriel\naria\n{claim}\n{ts.isoformat()}".encode()
886 bad_sig = encode_sig("ed25519", priv.sign(msign_msg))
887
888 req = AttestationRequest(
889 attester="gabriel", subject="aria", claim=claim,
890 signature=bad_sig, attester_public_key=pub, issued_at=ts,
891 )
892 with pytest.raises(ValueError, match="Invalid attestation signature"):
893 await issue_attestation(db_session, req)
894
895
896 @pytest.mark.tier7
897 @pytest.mark.asyncio
898 async def test_sec2_attester_impersonation_rejected(db_session: AsyncSession) -> None:
899 """Claiming attester=gabriel but signing with a different key → rejected."""
900 from musehub.services.musehub_attestations import issue_attestation
901
902 _, real_pub = _make_keypair() # gabriel's real key (pub only)
903 evil_priv, _ = _make_keypair() # attacker's key
904
905 ts = _utc()
906 claim = '{"type":"human"}'
907 # Attacker signs with own key but presents gabriel's pubkey
908 evil_sig = _sign(evil_priv, "gabriel", "aria", claim, ts.isoformat())
909
910 req = AttestationRequest(
911 attester="gabriel", subject="aria", claim=claim,
912 signature=evil_sig, attester_public_key=real_pub, issued_at=ts,
913 )
914 with pytest.raises(ValueError, match="Invalid attestation signature"):
915 await issue_attestation(db_session, req)
916
917
918 @pytest.mark.tier7
919 @pytest.mark.asyncio
920 async def test_sec3_tampered_claim_rejected(db_session: AsyncSession) -> None:
921 """Signature over original claim does not verify for a mutated claim."""
922 from musehub.services.musehub_attestations import issue_attestation
923
924 priv, pub = _make_keypair()
925 ts = _utc()
926 original_claim = '{"type":"collab"}'
927 tampered_claim = '{"type":"human"}' # different type
928 sig = _sign(priv, "gabriel", "aria", original_claim, ts.isoformat())
929
930 req = AttestationRequest(
931 attester="gabriel", subject="aria", claim=tampered_claim,
932 signature=sig, attester_public_key=pub, issued_at=ts,
933 )
934 with pytest.raises(ValueError, match="Invalid attestation signature"):
935 await issue_attestation(db_session, req)
936
937
938 @pytest.mark.tier7
939 @pytest.mark.asyncio
940 async def test_sec4_revoker_impersonation_rejected(db_session: AsyncSession) -> None:
941 """Wrong revoker cannot retract someone else's attestation."""
942 from musehub.services.musehub_attestations import issue_attestation, revoke_attestation
943
944 priv, pub = _make_keypair()
945 req = _req("gabriel", "aria", '{"type":"trusted"}', priv, pub)
946 result = await issue_attestation(db_session, req)
947
948 with pytest.raises(ValueError, match="only the attester can revoke"):
949 await revoke_attestation(db_session, result.attestation_id, revoker="evil-actor")
950
951
952 @pytest.mark.tier7
953 @pytest.mark.asyncio
954 async def test_sec5_expired_attestation_excluded_without_flag(db_session: AsyncSession) -> None:
955 """Expired attestation is excluded from live queries even if not explicitly revoked."""
956 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
957
958 priv, pub = _make_keypair()
959 past = _utc() - timedelta(hours=1)
960 req = _req("gabriel", "aria-exp", '{"type":"contractor"}', priv, pub, expires_at=past)
961 result = await issue_attestation(db_session, req)
962
963 listed = await get_attestations_for_subject(db_session, "aria-exp")
964 assert all(a.attestation_id != result.attestation_id for a in listed.attestations)
965 assert result.revoked_at is None # not revoked — just expired
966
967
968 @pytest.mark.tier7
969 @pytest.mark.asyncio
970 async def test_sec6_key_rotation_old_attestations_still_verifiable(db_session: AsyncSession) -> None:
971 """attester_public_key is stored per-attestation; key rotation doesn't break old records."""
972 from musehub.services.musehub_attestations import issue_attestation, verify_stored_attestation
973
974 old_priv, old_pub = _make_keypair()
975 req = _req("gabriel", "aria", '{"type":"collab"}', old_priv, old_pub)
976 result = await issue_attestation(db_session, req)
977
978 # Simulate key rotation — new key exists but old attestation used old key
979 _new_priv, _new_pub = _make_keypair()
980
981 # Verify using the stored public key (not the current key)
982 ok, reason = await verify_stored_attestation(db_session, result.attestation_id)
983 assert ok, f"old attestation failed verification after key rotation: {reason}"
984
985
986 @pytest.mark.tier7
987 @pytest.mark.asyncio
988 async def test_sec7_sql_metachar_in_claim_stored_safely(db_session: AsyncSession) -> None:
989 """SQL metacharacters in claim payload are stored verbatim without injection."""
990 from musehub.services.musehub_attestations import issue_attestation, get_attestations_for_subject
991
992 priv, pub = _make_keypair()
993 evil_claim = '{"type":"collab","note":"Robert\'); DROP TABLE musehub_attestations;--"}'
994 req = _req("gabriel", "aria-sql", evil_claim, priv, pub)
995 result = await issue_attestation(db_session, req)
996
997 listed = await get_attestations_for_subject(db_session, "aria-sql")
998 found = next((a for a in listed.attestations if a.attestation_id == result.attestation_id), None)
999 assert found is not None
1000 assert "DROP TABLE" in found.claim # stored verbatim, not executed
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago