tests/core/test_oidc.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """OIDC login method tests (core 1d wave 3 — provider, ``start``, ``callback``, e2e).16 17 Drives a full hand-built ``AsgiServer`` at the ASGI level (the ``test_login_flow``18 style): an OIDC provider is configured through the ``server_app`` config lift, so19 the ``ServerApplication`` registers one ``OidcMethod`` under20 ``/_server/auth/oidc:<code>/``. Both routes are exercised end to end — ``start``21 (the authorization redirect) and ``callback`` (code exchange, id_token22 validation, avatar attach) — plus a single closing e2e that runs the whole round23 trip. The external dependencies (discovery, the token endpoint, the JWKS) are24 mocked at the httpx and ``jwt.PyJWKClient`` boundaries; a module RSA keypair signs25 the id_token. The descriptor, ``login_methods`` ordering, the S256 PKCE challenge26 and the lazy-discovery boot guarantee are covered alongside.27 """28 29 from __future__ import annotations30 31 import base6432 import hashlib33 import json34 from typing import Any35 from urllib.parse import parse_qs, urlparse36 37 import httpx38 import jwt39 import pytest40 from cryptography.hazmat.primitives.asymmetric import rsa41 42 from genro_asgi import AsgiServer, BaseApplication, OidcMethod, ServerApplication43 from genro_asgi.types import Message, Scope44 45 DISCOVERY_DOC = {46 "authorization_endpoint": "https://accounts.example.com/authorize",47 "token_endpoint": "https://accounts.example.com/token",48 "jwks_uri": "https://accounts.example.com/jwks",49 }50 51 PROVIDER = {52 "issuer": "https://accounts.example.com",53 "client_id": "client-123",54 "scopes": "openid email profile",55 "identity_claim": "email",56 "tags": [],57 }58 59 # One RSA keypair for the whole module: the "provider" signs the id_token with the60 # private key, the callback verifies it with the public key served as the JWKS.61 _SIGNING_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048)62 63 64 def make_id_token(65 private_key: Any = _SIGNING_KEY,66 *,67 issuer: str = "https://accounts.example.com",68 audience: str = "client-123",69 claims: dict[str, Any] | None = None,70 ) -> str:71 """A signed RS256 id_token — the provider's response to the token exchange."""72 payload: dict[str, Any] = {"iss": issuer, "aud": audience, "email": "alice@example.com"}73 payload.update(claims or {})74 return jwt.encode(payload, private_key, algorithm="RS256")75 76 77 class _FakeSigningKey:78 """Stand-in for a ``jwt.PyJWK`` — exposes the public key as ``.key``."""79 80 def __init__(self, key: Any) -> None:81 self.key = key82 83 84 class _FakePyJWKClient:85 """Stand-in for ``jwt.PyJWKClient``: resolves every token to the module key.86 87 ``instances`` counts constructions: the real client caches the fetched key88 set on the instance, so building one per callback would re-fetch the JWKS89 every time — the count pins that the method keeps ONE.90 """91 92 instances = 093 94 def __init__(self, uri: str) -> None:95 type(self).instances += 196 97 def get_signing_key_from_jwt(self, token: str) -> _FakeSigningKey:98 return _FakeSigningKey(_SIGNING_KEY.public_key())99 100 101 class FakeResponse:102 """Minimal stand-in for an ``httpx.Response`` — carries a canned JSON payload."""103 104 def __init__(self, payload: dict[str, Any]) -> None:105 self._payload = payload106 107 def raise_for_status(self) -> None:108 pass109 110 def json(self) -> dict[str, Any]:111 return self._payload112 113 114 class FakeAsyncClient:115 """Stand-in for ``httpx.AsyncClient``: GET returns discovery, POST the token set.116 117 ``token_response`` (a class attribute swapped per test) is what the token118 endpoint answers; the default carries a freshly signed id_token. ``posted``119 records every token-exchange form so a test can assert what the provider120 actually received.121 """122 123 token_response: dict[str, Any] = {}124 posted: list[dict[str, Any]] = []125 126 def __init__(self, *args: Any, **kwargs: Any) -> None:127 pass128 129 async def __aenter__(self) -> FakeAsyncClient:130 return self131 132 async def __aexit__(self, *exc: Any) -> None:133 pass134 135 async def get(self, url: str) -> FakeResponse:136 return FakeResponse(DISCOVERY_DOC)137 138 async def post(self, url: str, data: dict[str, Any] | None = None) -> FakeResponse:139 type(self).posted.append(dict(data or {}))140 return FakeResponse(self.token_response or {"id_token": make_id_token()})141 142 143 @pytest.fixture144 def mock_discovery(monkeypatch: pytest.MonkeyPatch) -> None:145 """Mock the httpx boundary so ``discovery()`` returns the canned document."""146 monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)147 148 149 @pytest.fixture150 def mock_provider(monkeypatch: pytest.MonkeyPatch) -> None:151 """Mock discovery, the JWKS signing-key resolution, and the token endpoint."""152 FakeAsyncClient.token_response = {}153 FakeAsyncClient.posted = []154 _FakePyJWKClient.instances = 0155 monkeypatch.setattr(httpx, "AsyncClient", FakeAsyncClient)156 monkeypatch.setattr(jwt, "PyJWKClient", _FakePyJWKClient)157 158 159 EXTERNAL_URL = "https://shop.example.com"160 161 162 def make_server(provider: dict[str, Any] | None = None) -> AsgiServer:163 """A hand-built server carrying one OIDC provider under code ``google``.164 165 ``external_url`` is mandatory alongside a provider: it is the prefix of the166 absolute ``redirect_uri`` the provider is handed, so a server configured167 without it refuses to boot.168 """169 return AsgiServer(170 applications=[BaseApplication(mount="")],171 external_url=EXTERNAL_URL,172 server_app={"oidc": {"google": provider if provider is not None else PROVIDER}},173 )174 175 176 async def drive(177 server: AsgiServer,178 path: str,179 method: str = "GET",180 cookie: str | None = None,181 ) -> tuple[Scope, list[Message]]:182 """Drive one request through ``server`` at the ASGI level."""183 headers: list[tuple[bytes, bytes]] = []184 if cookie is not None:185 headers.append((b"cookie", cookie.encode()))186 path, _, query = path.partition("?")187 scope: Scope = {188 "type": "http",189 "method": method,190 "path": path,191 "query_string": query.encode(),192 "headers": headers,193 }194 sent: list[Message] = []195 196 async def receive() -> Message:197 return {"type": "http.request", "body": b"", "more_body": False}198 199 async def send(message: Message) -> None:200 sent.append(message)201 202 await server(scope, receive, send)203 return scope, sent204 205 206 def response_status(sent: list[Message]) -> int:207 return next(m["status"] for m in sent if m["type"] == "http.response.start")208 209 210 def response_headers(sent: list[Message]) -> dict[bytes, bytes]:211 start = next(m for m in sent if m["type"] == "http.response.start")212 return {name: value for name, value in start["headers"]}213 214 215 def response_body(sent: list[Message]) -> bytes:216 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")217 218 219 def json_body(sent: list[Message]) -> Any:220 return json.loads(response_body(sent))221 222 223 def location_query(sent: list[Message]) -> dict[str, list[str]]:224 """The parsed query of the redirect ``Location`` header."""225 location = response_headers(sent)[b"location"].decode()226 return parse_qs(urlparse(location).query)227 228 229 class TestOidcDescriptor:230 def test_descriptor_shape_and_no_secret_leak(self) -> None:231 server = make_server()232 app = server.applications["_server"]233 assert isinstance(app, ServerApplication)234 assert app.auth_section is not None235 method = app.auth_section.methods["oidc:google"]236 assert isinstance(method, OidcMethod)237 assert method.descriptor() == {238 "id": "oidc:google",239 "kind": "redirect",240 "label": "Sign in with google",241 "url": "/_server/auth/oidc:google/start",242 }243 # The public descriptor never carries the client id, issuer, or secret.244 for leak in ("client_id", "issuer", "client_secret"):245 assert leak not in method.descriptor()246 247 def test_pkce_challenge_matches_the_rfc_7636_vector(self) -> None:248 # RFC 7636 Appendix B's S256 example: a fixed verifier/challenge pair, so an249 # implementation that is wrong but self-consistent cannot pass.250 server = make_server()251 app = server.applications["_server"]252 assert isinstance(app, ServerApplication)253 assert app.auth_section is not None254 method = app.auth_section.methods["oidc:google"]255 assert isinstance(method, OidcMethod)256 assert (257 method._pkce_challenge("dBjftJeZ4CVP-mB92K27uhbUJU1p1r_wW1gFWFOEjXk")258 == "E9Melhoa2OwvFrEMTJguCHaoeK1t8URWbuGJSstw-cM"259 )260 261 def test_method_owns_routes_and_is_attached(self) -> None:262 server = make_server()263 app = server.applications["_server"]264 assert isinstance(app, ServerApplication)265 assert app.auth_section is not None266 routers = app.auth_section.route.nodes(lazy=True).get("routers") or {}267 assert "oidc:google" in routers268 269 async def test_login_methods_lists_oidc_after_password(self) -> None:270 server = make_server()271 _, sent = await drive(server, "/_server/login_methods")272 assert response_status(sent) == 200273 methods = json_body(sent)["methods"]274 assert [m["id"] for m in methods] == ["password", "oidc:google"]275 assert methods[1]["kind"] == "redirect"276 assert methods[1]["label"] == "Sign in with google"277 assert methods[1]["url"] == "/_server/auth/oidc:google/start"278 279 280 class TestOidcStart:281 async def test_start_redirects_to_the_authorization_endpoint(282 self, mock_discovery: None283 ) -> None:284 server = make_server()285 session = server.session_store.create()286 _, sent = await drive(287 server,288 "/_server/auth/oidc:google/start?next=/app/page",289 cookie=f"session_id={session.id}",290 )291 assert response_status(sent) == 302292 location = response_headers(sent)[b"location"].decode()293 assert location.startswith("https://accounts.example.com/authorize?")294 query = location_query(sent)295 assert query["response_type"] == ["code"]296 assert query["client_id"] == ["client-123"]297 # ABSOLUTE, built from the server's external_url (RFC 6749 §3.1.2): a298 # relative redirect_uri is refused by every real provider.299 assert query["redirect_uri"] == [300 f"{EXTERNAL_URL}/_server/auth/oidc:google/callback"301 ]302 assert query["scope"] == ["openid email profile"]303 assert query["code_challenge_method"] == ["S256"]304 assert query["state"][0]305 assert query["code_challenge"][0]306 307 async def test_start_lands_state_verifier_next_in_the_session(308 self, mock_discovery: None309 ) -> None:310 server = make_server()311 session = server.session_store.create()312 _, sent = await drive(313 server,314 "/_server/auth/oidc:google/start?next=/app/page",315 cookie=f"session_id={session.id}",316 )317 reloaded = server.session_store.get(session.id)318 assert reloaded is not None319 state = reloaded.data["oidc.oidc:google.state"]320 verifier = reloaded.data["oidc.oidc:google.verifier"]321 assert state and verifier322 assert reloaded.data["oidc.oidc:google.next"] == "/app/page"323 # The state minted for the session is the state carried in the redirect.324 assert location_query(sent)["state"] == [state]325 326 async def test_start_stores_only_a_safe_next(self, mock_discovery: None) -> None:327 server = make_server()328 session = server.session_store.create()329 await drive(330 server,331 "/_server/auth/oidc:google/start?next=//evil.example",332 cookie=f"session_id={session.id}",333 )334 reloaded = server.session_store.get(session.id)335 assert reloaded is not None336 assert reloaded.data["oidc.oidc:google.next"] == "/" # collapsed by safe_next_path337 338 async def test_pkce_challenge_matches_the_stored_verifier(339 self, mock_discovery: None340 ) -> None:341 server = make_server()342 session = server.session_store.create()343 _, sent = await drive(344 server,345 "/_server/auth/oidc:google/start",346 cookie=f"session_id={session.id}",347 )348 reloaded = server.session_store.get(session.id)349 assert reloaded is not None350 verifier = reloaded.data["oidc.oidc:google.verifier"]351 challenge = location_query(sent)["code_challenge"][0]352 expected = base64.urlsafe_b64encode(353 hashlib.sha256(verifier.encode("ascii")).digest()354 ).rstrip(b"=").decode("ascii")355 assert challenge == expected356 357 358 class TestOidcRedirectUri:359 """The ``redirect_uri`` contract: absolute, stable, and declared not derived.360 361 A provider matches the URI byte for byte against the one registered for the362 client and refuses anything relative, so these are the assertions that decide363 whether a real login can happen at all.364 """365 366 async def test_redirect_uri_is_identical_in_start_and_token_exchange(367 self, mock_provider: None368 ) -> None:369 # The provider compares the two: a mismatch between the authorization370 # request and the exchange is invalid_grant. The exchange has no request371 # to derive a host from, which is why both read the declared external_url.372 server = make_server()373 session_id, state = await begin_flow(server)374 _, sent = await drive(375 server,376 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",377 cookie=f"session_id={session_id}",378 )379 assert response_status(sent) == 302380 exchanged = FakeAsyncClient.posted[-1]["redirect_uri"]381 assert exchanged == f"{EXTERNAL_URL}/_server/auth/oidc:google/callback"382 383 async def test_redirect_uri_carries_a_scheme_and_host(self, mock_discovery: None) -> None:384 server = make_server()385 session = server.session_store.create()386 _, sent = await drive(387 server,388 "/_server/auth/oidc:google/start",389 cookie=f"session_id={session.id}",390 )391 parsed = urlparse(location_query(sent)["redirect_uri"][0])392 assert parsed.scheme == "https"393 assert parsed.netloc == "shop.example.com"394 assert parsed.path == "/_server/auth/oidc:google/callback"395 396 async def test_external_url_trailing_slash_does_not_double_the_separator(397 self, mock_discovery: None398 ) -> None:399 server = AsgiServer(400 applications=[BaseApplication(mount="")],401 external_url="https://shop.example.com/",402 server_app={"oidc": {"google": PROVIDER}},403 )404 session = server.session_store.create()405 _, sent = await drive(406 server,407 "/_server/auth/oidc:google/start",408 cookie=f"session_id={session.id}",409 )410 assert location_query(sent)["redirect_uri"] == [411 "https://shop.example.com/_server/auth/oidc:google/callback"412 ]413 414 def test_a_configured_provider_without_external_url_refuses_to_boot(self) -> None:415 # An incomplete configuration is loud at boot, not an opaque provider416 # error at the first login attempt.417 with pytest.raises(ValueError, match="external_url"):418 AsgiServer(419 applications=[BaseApplication(mount="")],420 server_app={"oidc": {"google": PROVIDER}},421 )422 423 def test_a_server_without_providers_needs_no_external_url(self) -> None:424 server = AsgiServer(applications=[BaseApplication(mount="")])425 assert server.external_url is None426 427 428 class TestOidcDiscoveryLazy:429 def test_construction_does_not_fetch_discovery(self) -> None:430 # An unreachable issuer must not break server construction: discovery is431 # lazy, never called in __init__ (the offline-boot guarantee).432 server = make_server({"issuer": "https://unreachable.invalid", "client_id": "x"})433 app = server.applications["_server"]434 assert isinstance(app, ServerApplication)435 assert app.auth_section is not None436 method = app.auth_section.methods["oidc:google"]437 assert isinstance(method, OidcMethod)438 assert method._discovery is None # nothing fetched at construction439 440 441 async def begin_flow(server: AsgiServer, next_target: str = "/app/page") -> tuple[str, str]:442 """Run ``start`` for a fresh session; return ``(session_id, minted_state)``.443 444 Lands the real ``state``/``verifier``/``next`` in the session exactly as a445 live authorization request would, so the callback exercises the true round446 trip rather than a hand-seeded facsimile.447 """448 session = server.session_store.create()449 await drive(450 server,451 f"/_server/auth/oidc:google/start?next={next_target}",452 cookie=f"session_id={session.id}",453 )454 reloaded = server.session_store.get(session.id)455 assert reloaded is not None456 return session.id, reloaded.data["oidc.oidc:google.state"]457 458 459 class TestOidcCallback:460 async def test_callback_authenticates_and_redirects_to_next(self, mock_provider: None) -> None:461 server = make_server()462 session_id, state = await begin_flow(server)463 _, sent = await drive(464 server,465 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",466 cookie=f"session_id={session_id}",467 )468 assert response_status(sent) == 302469 assert response_headers(sent)[b"location"].decode() == "/app/page"470 promoted = server.session_store.get(session_id)471 assert promoted is not None and promoted.avatar() is not None472 assert promoted.avatar().identity == "alice@example.com"473 assert promoted.avatar().tags == []474 # The session id never changes at login (no cookie involved).475 assert promoted.id == session_id476 477 async def test_callback_clears_the_one_shot_flow_keys(self, mock_provider: None) -> None:478 server = make_server()479 session_id, state = await begin_flow(server)480 await drive(481 server,482 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",483 cookie=f"session_id={session_id}",484 )485 promoted = server.session_store.get(session_id)486 assert promoted is not None487 for suffix in ("state", "verifier", "next"):488 assert f"oidc.oidc:google.{suffix}" not in promoted.data489 490 async def test_callback_does_not_touch_the_lockout_counter(self, mock_provider: None) -> None:491 # OIDC never routes through the password login route, so the store-backed492 # failure counter (Phase 2) is untouched by a successful OIDC login.493 server = make_server()494 session_id, state = await begin_flow(server)495 await drive(496 server,497 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",498 cookie=f"session_id={session_id}",499 )500 app = server.applications["_server"]501 assert isinstance(app, ServerApplication)502 user_store = getattr(server, "user_store", None)503 assert user_store is None # no user store wired: nothing to touch504 505 async def test_callback_rejects_a_mismatched_state(self, mock_provider: None) -> None:506 server = make_server()507 session_id, _ = await begin_flow(server)508 _, sent = await drive(509 server,510 "/_server/auth/oidc:google/callback?state=forged&code=auth-code",511 cookie=f"session_id={session_id}",512 )513 assert response_status(sent) == 400514 promoted = server.session_store.get(session_id)515 assert promoted is not None and promoted.avatar() is None516 517 async def test_callback_rejects_a_missing_state(self, mock_provider: None) -> None:518 server = make_server()519 session_id, _ = await begin_flow(server)520 _, sent = await drive(521 server,522 "/_server/auth/oidc:google/callback?code=auth-code",523 cookie=f"session_id={session_id}",524 )525 assert response_status(sent) == 400526 527 async def test_callback_rejects_a_token_response_without_id_token(528 self, monkeypatch: pytest.MonkeyPatch, mock_provider: None529 ) -> None:530 server = make_server()531 session_id, state = await begin_flow(server)532 monkeypatch.setattr(FakeAsyncClient, "token_response", {"access_token": "opaque"})533 _, sent = await drive(534 server,535 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",536 cookie=f"session_id={session_id}",537 )538 assert response_status(sent) == 400539 540 async def test_callback_rejects_a_wrong_audience_id_token(541 self, monkeypatch: pytest.MonkeyPatch, mock_provider: None542 ) -> None:543 server = make_server()544 session_id, state = await begin_flow(server)545 forged = make_id_token(audience="someone-else")546 monkeypatch.setattr(FakeAsyncClient, "token_response", {"id_token": forged})547 _, sent = await drive(548 server,549 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",550 cookie=f"session_id={session_id}",551 )552 assert response_status(sent) == 400553 promoted = server.session_store.get(session_id)554 assert promoted is not None and promoted.avatar() is None555 556 async def test_callback_rejects_a_wrong_issuer_id_token(557 self, monkeypatch: pytest.MonkeyPatch, mock_provider: None558 ) -> None:559 server = make_server()560 session_id, state = await begin_flow(server)561 forged = make_id_token(issuer="https://evil.example")562 monkeypatch.setattr(FakeAsyncClient, "token_response", {"id_token": forged})563 _, sent = await drive(564 server,565 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",566 cookie=f"session_id={session_id}",567 )568 assert response_status(sent) == 400569 570 async def test_callback_rejects_a_bad_signature_id_token(571 self, monkeypatch: pytest.MonkeyPatch, mock_provider: None572 ) -> None:573 # An id_token signed by a DIFFERENT key than the JWKS resolves — the574 # signature check must reject it.575 server = make_server()576 session_id, state = await begin_flow(server)577 other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)578 forged = make_id_token(private_key=other_key)579 monkeypatch.setattr(FakeAsyncClient, "token_response", {"id_token": forged})580 _, sent = await drive(581 server,582 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",583 cookie=f"session_id={session_id}",584 )585 assert response_status(sent) == 400586 587 async def test_callback_rejects_a_missing_identity_claim(588 self, monkeypatch: pytest.MonkeyPatch, mock_provider: None589 ) -> None:590 server = make_server()591 session_id, state = await begin_flow(server)592 no_email = make_id_token(claims={"email": None})593 monkeypatch.setattr(FakeAsyncClient, "token_response", {"id_token": no_email})594 _, sent = await drive(595 server,596 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",597 cookie=f"session_id={session_id}",598 )599 assert response_status(sent) == 400600 601 602 class TestOidcJwksResolution:603 """How the id_token's signing key is fetched: once, and off the event loop."""604 605 async def test_the_jwks_client_is_built_once_across_callbacks(606 self, mock_provider: None607 ) -> None:608 # PyJWKClient caches the key set ON THE INSTANCE, so a client per callback609 # would re-fetch the JWKS on every login. One instance serves the method.610 server = make_server()611 for _ in range(2):612 session_id, state = await begin_flow(server)613 _, sent = await drive(614 server,615 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",616 cookie=f"session_id={session_id}",617 )618 assert response_status(sent) == 302619 assert _FakePyJWKClient.instances == 1620 621 async def test_the_signing_key_is_resolved_off_the_event_loop(622 self, mock_provider: None623 ) -> None:624 # The real client fetches with a blocking stdlib call: it must run on the625 # server pool (D2), never on the loop. The pool is provisioned lazily, so626 # having been provisioned by the callback is the observable proof.627 server = make_server()628 assert server.pool.provisioned is False629 session_id, state = await begin_flow(server)630 _, sent = await drive(631 server,632 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",633 cookie=f"session_id={session_id}",634 )635 assert response_status(sent) == 302636 assert server.pool.provisioned is True637 638 639 class TestOidcEndToEnd:640 """The closing barrier: one configured provider, discovered → started →641 called back → authenticated, over the real ASGI drive — the whole round trip642 a browser makes, in a single narrative."""643 644 async def test_configured_provider_logs_a_user_in_end_to_end(645 self, mock_provider: None646 ) -> None:647 server = make_server()648 649 # 1. The provider appears as a redirect descriptor in login_methods.650 _, sent = await drive(server, "/_server/login_methods")651 methods = json_body(sent)["methods"]652 oidc = next(m for m in methods if m["id"] == "oidc:google")653 assert oidc["kind"] == "redirect"654 assert oidc["url"] == "/_server/auth/oidc:google/start"655 656 # 2. start: the browser is redirected to the provider's authorize endpoint.657 session = server.session_store.create()658 cookie = f"session_id={session.id}"659 _, sent = await drive(server, oidc["url"] + "?next=/app/home", cookie=cookie)660 assert response_status(sent) == 302661 authorize = response_headers(sent)[b"location"].decode()662 assert authorize.startswith("https://accounts.example.com/authorize?")663 query = location_query(sent)664 state = query["state"][0]665 666 # 3. callback: the provider redirects back; the code is exchanged, the667 # id_token validated, the avatar attached, the browser sent to next.668 _, sent = await drive(669 server,670 f"/_server/auth/oidc:google/callback?state={state}&code=auth-code",671 cookie=cookie,672 )673 assert response_status(sent) == 302674 assert response_headers(sent)[b"location"].decode() == "/app/home"675 676 # 4. The session now carries the authenticated identity (id unchanged),677 # and the OIDC login never touched the password lockout machinery.678 authenticated = server.session_store.get(session.id)679 assert authenticated is not None and authenticated.avatar() is not None680 assert authenticated.avatar().identity == "alice@example.com"681 assert authenticated.id == session.id682 assert getattr(server, "user_store", None) is None