gabriel / musehub public
test_agent_registration.py python
457 lines 16.3 KB
Raw
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ breaking 143 days ago
1 """Tests for agent identity provisioning.
2
3 Covers:
4 - AgentRegistrationRequest model validation
5 - register_agent_identity service function (unit tests with mocked DB)
6 - POST /api/identities/agent route (happy path + error cases)
7 - verify_and_authenticate identity_type support
8 - VerifyRequest identity_type field validation
9 """
10 from __future__ import annotations
11
12 import uuid
13 from unittest.mock import AsyncMock, MagicMock, patch
14
15 import pytest
16 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
17 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
18 from httpx import AsyncClient
19 from muse.core.types import encode_pubkey, public_key_fingerprint
20
21 from musehub.types.json_types import StrDict
22 from musehub.models.musehub_auth import (
23 AgentRegistrationRequest,
24 AgentRegistrationResponse,
25 VerifyRequest,
26 )
27
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33
34 def _generate_key_material() -> tuple[str, str]:
35 """Generate a fresh Ed25519 keypair and return (public_key_b64, fingerprint)."""
36 key = Ed25519PrivateKey.generate()
37 pub_raw = key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
38 return encode_pubkey("ed25519", pub_raw), public_key_fingerprint(pub_raw)
39
40
41 # ---------------------------------------------------------------------------
42 # AgentRegistrationRequest validation
43 # ---------------------------------------------------------------------------
44
45
46 class TestAgentRegistrationRequestValidation:
47 def _valid_payload(self) -> JSONObject:
48 pub_b64, fp = _generate_key_material()
49 return {
50 "handle": "agentception-abc123",
51 "public_key_b64": pub_b64,
52 "fingerprint": fp,
53 "algorithm": "ed25519",
54 "agent_model": "claude-sonnet-4-6",
55 "scope": ["push:agentception"],
56 "label": "ephemeral/agentception-abc123",
57 }
58
59 def test_valid_request_parses(self) -> None:
60 req = AgentRegistrationRequest(**self._valid_payload())
61 assert req.handle == "agentception-abc123"
62 assert req.algorithm == "ed25519"
63 assert req.agent_model == "claude-sonnet-4-6"
64 assert req.scope == ["push:agentception"]
65
66 def test_handle_normalised_to_lowercase(self) -> None:
67 payload = self._valid_payload()
68 payload["handle"] = "AgentCeption-ABC"
69 req = AgentRegistrationRequest(**payload)
70 assert req.handle == "agentception-abc"
71
72 def test_invalid_handle_rejected(self) -> None:
73 payload = self._valid_payload()
74 payload["handle"] = "agent with spaces"
75 with pytest.raises(Exception): # ValidationError
76 AgentRegistrationRequest(**payload)
77
78 def test_fingerprint_must_be_64_hex(self) -> None:
79 payload = self._valid_payload()
80 payload["fingerprint"] = "tooshort"
81 with pytest.raises(Exception):
82 AgentRegistrationRequest(**payload)
83
84 def test_invalid_algorithm_rejected(self) -> None:
85 payload = self._valid_payload()
86 payload["algorithm"] = "rsa-2048"
87 with pytest.raises(Exception):
88 AgentRegistrationRequest(**payload)
89
90 def test_scope_defaults_to_empty_list(self) -> None:
91 payload = self._valid_payload()
92 del payload["scope"]
93 req = AgentRegistrationRequest(**payload)
94 assert req.scope == []
95
96 def test_expires_at_defaults_to_none(self) -> None:
97 req = AgentRegistrationRequest(**self._valid_payload())
98 assert req.expires_at is None
99
100 def test_expires_at_accepts_iso8601(self) -> None:
101 payload = self._valid_payload()
102 payload["expires_at"] = "2026-04-06T14:00:00Z"
103 req = AgentRegistrationRequest(**payload)
104 assert req.expires_at == "2026-04-06T14:00:00Z"
105
106
107 # ---------------------------------------------------------------------------
108 # VerifyRequest identity_type field
109 # ---------------------------------------------------------------------------
110
111
112 class TestVerifyRequestIdentityType:
113 def _base_payload(self) -> JSONObject:
114 return {
115 "challenge_token": "a" * 64,
116 "public_key_b64": "AAEC",
117 "signature_b64": "AAEC",
118 }
119
120 def test_default_identity_type_is_human(self) -> None:
121 req = VerifyRequest(**self._base_payload())
122 assert req.identity_type == "human"
123
124 def test_agent_identity_type_accepted(self) -> None:
125 req = VerifyRequest(**self._base_payload(), identity_type="agent")
126 assert req.identity_type == "agent"
127
128 def test_invalid_identity_type_rejected(self) -> None:
129 with pytest.raises(Exception):
130 VerifyRequest(**self._base_payload(), identity_type="robot")
131
132
133 # ---------------------------------------------------------------------------
134 # register_agent_identity service — unit tests
135 # ---------------------------------------------------------------------------
136
137
138 class TestRegisterAgentIdentityService:
139 """Unit tests using a mock DB session — no real DB required."""
140
141 @staticmethod
142 def _make_created_at() -> None:
143 from datetime import datetime, timezone
144 t = MagicMock()
145 t.isoformat.return_value = "2026-04-06T00:00:00+00:00"
146 return t
147
148 def _make_mock_session(
149 self, *, key_row_exists: bool = False, identity_row_exists: bool = False
150 ) -> AsyncMock:
151 session = AsyncMock()
152
153 # Simulated scalar_one_or_none return for SELECT MusehubAuthKey
154 mock_scalar_result = MagicMock()
155 if key_row_exists:
156 mock_key = MagicMock()
157 mock_key.key_id = str(uuid.uuid4())
158 mock_key.identity_id = str(uuid.uuid4())
159 mock_key.algorithm = "ed25519"
160 mock_key.fingerprint = "a" * 64
161 mock_key.label = "existing"
162 mock_key.created_at = self._make_created_at()
163 mock_key.last_used_at = None
164 mock_scalar_result.scalar_one_or_none.return_value = mock_key
165
166 if identity_row_exists:
167 mock_identity = MagicMock()
168 mock_identity.identity_id = mock_key.identity_id
169 mock_identity.handle = "agentception-abc"
170 # second execute call returns identity
171 mock_scalar_result2 = MagicMock()
172 mock_scalar_result2.scalar_one_or_none.return_value = mock_identity
173 session.execute.side_effect = [
174 mock_scalar_result,
175 mock_scalar_result2,
176 ]
177 else:
178 session.execute.return_value = mock_scalar_result
179 else:
180 mock_scalar_result.scalar_one_or_none.return_value = None
181 session.execute.return_value = mock_scalar_result
182
183 session.flush = AsyncMock()
184 session.commit = AsyncMock()
185 session.add = MagicMock()
186
187 # refresh populates created_at on any ORM object passed to it
188 async def _mock_refresh(obj: MagicMock) -> None:
189 if not hasattr(obj, "created_at") or obj.created_at is None:
190 obj.created_at = self._make_created_at()
191 if not hasattr(obj, "last_used_at"):
192 obj.last_used_at = None
193
194 session.refresh = _mock_refresh
195
196 return session
197
198 @pytest.mark.asyncio
199 async def test_new_agent_registration_creates_identity_and_key(self) -> None:
200 from musehub.services.musehub_auth import register_agent_identity
201
202 pub_b64, fp = _generate_key_material()
203 session = self._make_mock_session(key_row_exists=False)
204
205 result = await register_agent_identity(
206 session=session,
207 handle="agentception-abc",
208 public_key_b64=pub_b64,
209 fingerprint=fp,
210 algorithm="ed25519",
211 spawned_by="gabriel",
212 agent_model="claude-sonnet-4-6",
213 scope=["push:agentception"],
214 )
215
216 assert result.is_new_identity is True
217 assert result.spawned_by == "gabriel"
218 assert result.handle == "agentception-abc"
219 # session.add called twice: identity + key
220 assert session.add.call_count == 2
221 assert session.commit.call_count == 1
222
223 @pytest.mark.asyncio
224 async def test_fingerprint_mismatch_raises_auth_error(self) -> None:
225 from musehub.services.musehub_auth import AuthError, register_agent_identity
226
227 pub_b64, _ = _generate_key_material()
228 wrong_fp = f"sha256:{'f' * 64}" # doesn't match the key
229
230 session = self._make_mock_session(key_row_exists=False)
231
232 with pytest.raises(AuthError) as exc_info:
233 await register_agent_identity(
234 session=session,
235 handle="agent-x",
236 public_key_b64=pub_b64,
237 fingerprint=wrong_fp,
238 algorithm="ed25519",
239 spawned_by="gabriel",
240 )
241
242 assert exc_info.value.status_code == 422
243 assert "fingerprint" in exc_info.value.detail.lower()
244
245 @pytest.mark.asyncio
246 async def test_invalid_public_key_b64_raises_auth_error(self) -> None:
247 from musehub.services.musehub_auth import AuthError, register_agent_identity
248
249 session = self._make_mock_session(key_row_exists=False)
250
251 with pytest.raises(AuthError) as exc_info:
252 await register_agent_identity(
253 session=session,
254 handle="agent-x",
255 public_key_b64="!!!not-base64!!!",
256 fingerprint=f"sha256:{'a' * 64}",
257 algorithm="ed25519",
258 spawned_by="gabriel",
259 )
260
261 assert exc_info.value.status_code == 422
262
263 @pytest.mark.asyncio
264 async def test_invalid_expires_at_raises_auth_error(self) -> None:
265 from musehub.services.musehub_auth import AuthError, register_agent_identity
266
267 pub_b64, fp = _generate_key_material()
268 session = self._make_mock_session(key_row_exists=False)
269
270 with pytest.raises(AuthError) as exc_info:
271 await register_agent_identity(
272 session=session,
273 handle="agent-x",
274 public_key_b64=pub_b64,
275 fingerprint=fp,
276 algorithm="ed25519",
277 spawned_by="gabriel",
278 expires_at="not-a-date",
279 )
280
281 assert exc_info.value.status_code == 422
282 assert "expires_at" in exc_info.value.detail.lower()
283
284 @pytest.mark.asyncio
285 async def test_expires_at_none_is_accepted(self) -> None:
286 from musehub.services.musehub_auth import register_agent_identity
287
288 pub_b64, fp = _generate_key_material()
289 session = self._make_mock_session(key_row_exists=False)
290
291 result = await register_agent_identity(
292 session=session,
293 handle="agent-no-expiry",
294 public_key_b64=pub_b64,
295 fingerprint=fp,
296 algorithm="ed25519",
297 spawned_by="gabriel",
298 expires_at=None,
299 )
300 assert result.is_new_identity is True
301
302
303 # ---------------------------------------------------------------------------
304 # POST /api/identities/agent route — HTTP integration tests
305 # ---------------------------------------------------------------------------
306
307
308 class TestProvisionAgentRoute:
309 """Integration tests using the full FastAPI test client."""
310
311 def _valid_payload(self) -> JSONObject:
312 pub_b64, fp = _generate_key_material()
313 return {
314 "handle": f"agent-{uuid.uuid4().hex[:8]}",
315 "public_key_b64": pub_b64,
316 "fingerprint": fp,
317 "algorithm": "ed25519",
318 "agent_model": "claude-sonnet-4-6",
319 "scope": ["push"],
320 "label": "test-ephemeral",
321 }
322
323 @pytest.mark.asyncio
324 async def test_provision_agent_happy_path(
325 self,
326 client: AsyncClient,
327 auth_headers: StrDict,
328 db_session: AsyncSession,
329 ) -> None:
330 payload = self._valid_payload()
331 resp = await client.post(
332 "/api/identities/agent",
333 json=payload,
334 headers=auth_headers,
335 )
336 assert resp.status_code in (200, 201)
337 data = resp.json()
338 assert data["handle"] == payload["handle"]
339 assert data["spawned_by"] == "testuser" # from auth_headers fixture
340 assert data["is_new_identity"] is True
341 assert "key" in data
342 assert data["key"]["algorithm"] == "ed25519"
343
344 @pytest.mark.asyncio
345 async def test_provision_agent_requires_auth(
346 self,
347 client: AsyncClient,
348 ) -> None:
349 payload = self._valid_payload()
350 resp = await client.post(
351 "/api/identities/agent",
352 json=payload,
353 # No auth headers
354 )
355 assert resp.status_code == 401
356
357 @pytest.mark.asyncio
358 async def test_provision_agent_duplicate_handle_409(
359 self,
360 client: AsyncClient,
361 auth_headers: StrDict,
362 db_session: AsyncSession,
363 ) -> None:
364 payload = self._valid_payload()
365 # First registration
366 r1 = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
367 assert r1.status_code in (200, 201)
368
369 # Same handle with a different key — should 409
370 _, fp2 = _generate_key_material()
371 pub_b64_2, fp2 = _generate_key_material()
372 payload2 = {**payload, "public_key_b64": pub_b64_2, "fingerprint": fp2}
373 r2 = await client.post("/api/identities/agent", json=payload2, headers=auth_headers)
374 assert r2.status_code == 409
375
376 @pytest.mark.asyncio
377 async def test_provision_agent_idempotent_same_key(
378 self,
379 client: AsyncClient,
380 auth_headers: StrDict,
381 db_session: AsyncSession,
382 ) -> None:
383 """Registering the same key twice is idempotent (returns 200 on re-register)."""
384 payload = self._valid_payload()
385 r1 = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
386 assert r1.status_code in (200, 201)
387
388 r2 = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
389 assert r2.status_code == 200
390 data2 = r2.json()
391 assert data2["is_new_identity"] is False
392
393 @pytest.mark.asyncio
394 async def test_provision_agent_invalid_handle_422(
395 self,
396 client: AsyncClient,
397 auth_headers: StrDict,
398 ) -> None:
399 payload = self._valid_payload()
400 payload["handle"] = "handle with spaces"
401 resp = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
402 assert resp.status_code == 422
403
404 @pytest.mark.asyncio
405 async def test_provision_agent_spawned_by_matches_operator(
406 self,
407 client: AsyncClient,
408 auth_headers: StrDict,
409 db_session: AsyncSession,
410 ) -> None:
411 payload = self._valid_payload()
412 resp = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
413 assert resp.status_code in (200, 201)
414 data = resp.json()
415 # _TEST_HANDLE from conftest
416 assert data["spawned_by"] == "testuser"
417
418 @pytest.mark.asyncio
419 async def test_provision_agent_scope_stored(
420 self,
421 client: AsyncClient,
422 auth_headers: StrDict,
423 db_session: AsyncSession,
424 ) -> None:
425 payload = self._valid_payload()
426 payload["scope"] = ["push:agentception", "pull:agentception"]
427 resp = await client.post("/api/identities/agent", json=payload, headers=auth_headers)
428 assert resp.status_code in (200, 201)
429
430
431 # ---------------------------------------------------------------------------
432 # verify_and_authenticate identity_type — unit test
433 # ---------------------------------------------------------------------------
434
435
436 class TestVerifyAuthenticateIdentityType:
437 """Unit test the identity_type param is propagated to the DB row."""
438
439 @pytest.mark.asyncio
440 async def test_identity_type_human_is_default(self) -> None:
441 """When identity_type is omitted it defaults to 'human'."""
442 req = VerifyRequest(
443 challenge_token="a" * 64,
444 public_key_b64="AAEC",
445 signature_b64="AAEC",
446 )
447 assert req.identity_type == "human"
448
449 @pytest.mark.asyncio
450 async def test_identity_type_agent_passes_through(self) -> None:
451 req = VerifyRequest(
452 challenge_token="a" * 64,
453 public_key_b64="AAEC",
454 signature_b64="AAEC",
455 identity_type="agent",
456 )
457 assert req.identity_type == "agent"
File History 1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a feat(intel): standardize headers, gauge icon, velocity card… Sonnet 4.6 minor ⚠ 143 days ago