import base64 import unittest from dataclasses import replace from time import time from typing import Any, Mapping, Optional from urllib.parse import parse_qs, urlparse import httpx import jwt from cryptography.hazmat.primitives.asymmetric import rsa from pwo import async_test from kaya.core import HttpContext, KayaApp from kaya.session import InMemorySessionStore, SessionMixin from kaya.oidc import OIDCConfig, OIDCMixin, OIDCProviders from kaya.oidc._client import OIDCClient class MockOIDCProvider: def __init__( self, name: str = 'mock', *, issuer: Optional[str] = None, sub: str = 'user123', email: str = 'user@example.com', metadata_issuer: Optional[str] = None, supports_par: bool = False, require_par: bool = False, par_error: bool = False, ) -> None: self.private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) self.public_key = self.private_key.public_key() self.kid = 'mock-key' self.issuer = issuer or f'https://{name}-oidc.local' self.metadata_issuer = metadata_issuer if metadata_issuer is not None else self.issuer self.sub = sub self.email = email self.nonce = f'{name}-nonce-value' self.token_issuer: Optional[str] = None self.token_tid: Optional[str] = None self.token_sub: Optional[str] = None self.token_expired = False self.par_error = par_error self.par_requests: list[dict[str, list[str]]] = [] self.par_auth_headers: list[Optional[str]] = [] self.discovery = { 'issuer': self.metadata_issuer, 'authorization_endpoint': f'{self.issuer}/auth', 'token_endpoint': f'{self.issuer}/token', 'userinfo_endpoint': f'{self.issuer}/userinfo', 'end_session_endpoint': f'{self.issuer}/logout', 'jwks_uri': f'{self.issuer}/jwks', } if supports_par or require_par: self.discovery['pushed_authorization_request_endpoint'] = f'{self.issuer}/par' if require_par: self.discovery['require_pushed_authorization_requests'] = True def _public_key_to_jwk(self) -> dict[str, str]: numbers = self.public_key.public_numbers() e_bytes = numbers.e.to_bytes((numbers.e.bit_length() + 7) // 8, 'big') n_bytes = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, 'big') return { 'kty': 'RSA', 'kid': self.kid, 'use': 'sig', 'n': base64.urlsafe_b64encode(n_bytes).rstrip(b'=').decode(), 'e': base64.urlsafe_b64encode(e_bytes).rstrip(b'=').decode(), 'alg': 'RS256', } def issue_id_token( self, nonce: Optional[str], audience: str, expired: bool = False, wrong_nonce: bool = False, issuer: Optional[str] = None, tid: Optional[str] = None, sub: Optional[str] = None, ) -> str: now = time() exp = now - 3600 if expired else now + 3600 payload: dict[str, Any] = { 'sub': sub if sub is not None else self.sub, 'iss': issuer if issuer is not None else self.issuer, 'aud': audience, 'iat': now, 'exp': exp, } if nonce is not None: payload['nonce'] = 'wrong-nonce' if wrong_nonce else nonce if tid is not None: payload['tid'] = tid return jwt.encode( payload, self.private_key, algorithm='RS256', headers={'kid': self.kid}, ) def issue_unsigned_id_token( self, nonce: str, audience: str, issuer: Optional[str] = None, ) -> str: payload: dict[str, Any] = { 'sub': self.sub, 'iss': issuer if issuer is not None else self.issuer, 'aud': audience, 'iat': time(), 'exp': time() + 3600, 'nonce': nonce, } return jwt.encode(payload, None, algorithm='none') def _issue_configured_token(self, audience: str) -> str: return self.issue_id_token( self.nonce, audience, expired=self.token_expired, issuer=self.token_issuer, tid=self.token_tid, sub=self.token_sub if self.token_sub is not None else self.sub, ) def handle_request(self, request: httpx.Request) -> httpx.Response: url = str(request.url) path = urlparse(url).path # Serve discovery and JWKS for any host so that issuer-mismatch tests can still fetch keys. if path.endswith('/.well-known/openid-configuration'): return httpx.Response(200, json=self.discovery) if path.endswith('/jwks'): return httpx.Response(200, json={'keys': [self._public_key_to_jwk()]}) if path == '/par': if request.method != 'POST': return httpx.Response(405) self.par_requests.append(parse_qs(request.content.decode())) self.par_auth_headers.append(request.headers.get('Authorization')) if self.par_error: return httpx.Response(400, json={ 'error': 'invalid_request', 'error_description': 'mock PAR failure', }) return httpx.Response(201, json={ 'request_uri': f'urn:ietf:params:oauth:request_uri:par-{len(self.par_requests)}', 'expires_in': 60, }) if path == '/token' and request.method == 'POST': body = request.content.decode() data = dict(part.split('=') for part in body.split('&')) if body else {} if data.get('grant_type') == 'refresh_token': return httpx.Response(200, json={ 'access_token': 'new-access-token', 'refresh_token': 'new-refresh-token', 'id_token': self._issue_configured_token(data.get('client_id', 'client')), 'token_type': 'Bearer', }) return httpx.Response(200, json={ 'access_token': 'mock-access-token', 'refresh_token': 'mock-refresh-token', 'id_token': self._issue_configured_token(data.get('client_id', 'client')), 'token_type': 'Bearer', }) if path == '/userinfo': return httpx.Response(200, json={'sub': self.sub, 'email': self.email}) if path == '/logout': return httpx.Response(200) return httpx.Response(404) class MockTransport(httpx.AsyncBaseTransport): def __init__(self, provider: MockOIDCProvider) -> None: self._provider = provider async def handle_async_request(self, request: httpx.Request) -> httpx.Response: return self._provider.handle_request(request) class OIDCClientTest(unittest.TestCase): def setUp(self) -> None: self.provider = MockOIDCProvider() self.http_client = httpx.AsyncClient(transport=MockTransport(self.provider)) self.config = OIDCConfig( issuer=self.provider.issuer, client_id='client', client_secret='secret', redirect_uri='http://localhost:8000/auth/callback', http_client=self.http_client, ) self.client = OIDCClient(self.config) @async_test async def test_build_authorization_url(self) -> None: url, state, nonce, verifier = await self.client.build_authorization_url() self.assertIn(self.provider.discovery['authorization_endpoint'], url) parsed = urlparse(url) query = parse_qs(parsed.query) self.assertEqual(['client'], query['client_id']) self.assertEqual(['code'], query['response_type']) self.assertEqual(['S256'], query['code_challenge_method']) self.assertIn('openid', query['scope'][0]) self.assertEqual([state], query['state']) self.assertEqual([nonce], query['nonce']) self.assertTrue(len(verifier) > 0) @async_test async def test_validate_id_token_success(self) -> None: token = self.provider.issue_id_token('nonce-value', 'client') claims = await self.client.validate_id_token(token, 'nonce-value') self.assertEqual('user123', claims['sub']) self.assertEqual('nonce-value', claims['nonce']) @async_test async def test_validate_id_token_wrong_nonce(self) -> None: token = self.provider.issue_id_token('nonce-value', 'client', wrong_nonce=True) with self.assertRaises(ValueError) as ctx: await self.client.validate_id_token(token, 'nonce-value') self.assertIn('nonce', str(ctx.exception)) @async_test async def test_validate_id_token_expired(self) -> None: token = self.provider.issue_id_token('nonce-value', 'client', expired=True) with self.assertRaises(jwt.ExpiredSignatureError): await self.client.validate_id_token(token, 'nonce-value') @async_test async def test_validate_id_token_wrong_issuer(self) -> None: provider = MockOIDCProvider(metadata_issuer='https://other-issuer.local') token = provider.issue_id_token('nonce-value', 'client') config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=httpx.AsyncClient(transport=MockTransport(provider)), ) client = OIDCClient(config) with self.assertRaises(jwt.InvalidIssuerError): await client.validate_id_token(token, 'nonce-value') @async_test async def test_validate_id_token_uses_discovered_issuer(self) -> None: # The configured issuer is only used as the discovery URL; the ID token # is validated against the issuer advertised by the discovery document. provider = MockOIDCProvider(metadata_issuer='https://canonical-issuer.local') token = provider.issue_id_token('nonce-value', 'client', issuer='https://canonical-issuer.local') config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=httpx.AsyncClient(transport=MockTransport(provider)), ) client = OIDCClient(config) claims = await client.validate_id_token(token, 'nonce-value') self.assertEqual('user123', claims['sub']) @async_test async def test_validate_id_token_entra_style_issuer_template(self) -> None: # Microsoft Entra ID multi-tenant metadata advertises # https://login.microsoftonline.com/{tenantid}/v2.0 and tokens carry # the concrete tenant id in both iss and tid. tenant = '11111111-2222-3333-4444-555555555555' provider = MockOIDCProvider(metadata_issuer='https://mock-oidc.local/{tenantid}/v2.0') token = provider.issue_id_token( 'nonce-value', 'client', issuer=f'https://mock-oidc.local/{tenant}/v2.0', tid=tenant, ) config = OIDCConfig( issuer=f'{provider.issuer}/common/v2.0', client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=httpx.AsyncClient(transport=MockTransport(provider)), ) client = OIDCClient(config) claims = await client.validate_id_token(token, 'nonce-value') self.assertEqual(tenant, claims['tid']) @async_test async def test_validate_id_token_template_with_wrong_tid(self) -> None: tenant = '11111111-2222-3333-4444-555555555555' other_tenant = '99999999-8888-7777-6666-555555555555' provider = MockOIDCProvider(metadata_issuer='https://mock-oidc.local/{tenantid}/v2.0') token = provider.issue_id_token( 'nonce-value', 'client', issuer=f'https://mock-oidc.local/{tenant}/v2.0', tid=other_tenant, ) config = OIDCConfig( issuer=f'{provider.issuer}/common/v2.0', client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=httpx.AsyncClient(transport=MockTransport(provider)), ) client = OIDCClient(config) with self.assertRaises(jwt.InvalidIssuerError): await client.validate_id_token(token, 'nonce-value') @async_test async def test_validate_id_token_template_without_tid(self) -> None: tenant = '11111111-2222-3333-4444-555555555555' provider = MockOIDCProvider(metadata_issuer='https://mock-oidc.local/{tenantid}/v2.0') token = provider.issue_id_token( 'nonce-value', 'client', issuer=f'https://mock-oidc.local/{tenant}/v2.0', ) config = OIDCConfig( issuer=f'{provider.issuer}/common/v2.0', client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=httpx.AsyncClient(transport=MockTransport(provider)), ) client = OIDCClient(config) with self.assertRaises(jwt.InvalidIssuerError): await client.validate_id_token(token, 'nonce-value') @async_test async def test_validate_refreshed_id_token_allows_missing_nonce(self) -> None: token = self.provider.issue_id_token(None, 'client') claims = await self.client.validate_refreshed_id_token(token, 'nonce-value') self.assertEqual('user123', claims['sub']) @async_test async def test_validate_refreshed_id_token_wrong_nonce(self) -> None: token = self.provider.issue_id_token('other-nonce', 'client') with self.assertRaises(ValueError) as ctx: await self.client.validate_refreshed_id_token(token, 'nonce-value') self.assertIn('nonce', str(ctx.exception)) @async_test async def test_validate_id_token_disallowed_algorithm(self) -> None: token = self.provider.issue_id_token('nonce-value', 'client') config = OIDCConfig( issuer=self.provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=self.http_client, allowed_id_token_algorithms=('ES256',), ) client = OIDCClient(config) with self.assertRaises(ValueError) as ctx: await client.validate_id_token(token, 'nonce-value') self.assertIn('RS256', str(ctx.exception)) @async_test async def test_validate_id_token_unsigned_allowed(self) -> None: token = self.provider.issue_unsigned_id_token('nonce-value', 'client') config = OIDCConfig( issuer=self.provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=self.http_client, allow_unsigned_id_tokens=True, ) client = OIDCClient(config) claims = await client.validate_id_token(token, 'nonce-value') self.assertEqual('user123', claims['sub']) @async_test async def test_validate_id_token_unsigned_rejected_by_default(self) -> None: token = self.provider.issue_unsigned_id_token('nonce-value', 'client') with self.assertRaises(ValueError) as ctx: await self.client.validate_id_token(token, 'nonce-value') self.assertIn('Unsigned', str(ctx.exception)) @async_test async def test_validate_id_token_unsigned_wrong_audience(self) -> None: token = self.provider.issue_unsigned_id_token('nonce-value', 'other-client') config = OIDCConfig( issuer=self.provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=self.http_client, allow_unsigned_id_tokens=True, ) client = OIDCClient(config) with self.assertRaises(jwt.InvalidAudienceError): await client.validate_id_token(token, 'nonce-value') @async_test async def test_validate_id_token_unsigned_wrong_issuer(self) -> None: token = self.provider.issue_unsigned_id_token('nonce-value', 'client', issuer='https://evil-issuer.local') config = OIDCConfig( issuer=self.provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=self.http_client, allow_unsigned_id_tokens=True, ) client = OIDCClient(config) with self.assertRaises(jwt.InvalidIssuerError): await client.validate_id_token(token, 'nonce-value') @async_test async def test_fetch_userinfo(self) -> None: userinfo = await self.client.fetch_userinfo('access-token') self.assertEqual('user123', userinfo['sub']) self.assertEqual('user@example.com', userinfo['email']) @async_test async def test_build_logout_url(self) -> None: url = await self.client.build_logout_url('id-token') self.assertIsNotNone(url) assert url is not None self.assertIn(self.provider.discovery['end_session_endpoint'], url) parsed = urlparse(url) query = parse_qs(parsed.query) self.assertEqual(['id-token'], query['id_token_hint']) self.assertEqual(['/'], query['post_logout_redirect_uri']) @async_test async def test_build_authorization_url_with_par(self) -> None: provider = MockOIDCProvider('par', supports_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', client_secret='secret', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, par='always', ) client = OIDCClient(config) url, state, nonce, verifier = await client.build_authorization_url(par='always') query = parse_qs(urlparse(url).query) self.assertEqual(['client'], query['client_id']) self.assertEqual(['urn:ietf:params:oauth:request_uri:par-1'], query['request_uri']) self.assertNotIn('state', query) self.assertNotIn('nonce', query) self.assertNotIn('code_challenge', query) self.assertEqual(1, len(provider.par_requests)) data = provider.par_requests[0] self.assertEqual(['client'], data['client_id']) self.assertEqual(['code'], data['response_type']) self.assertEqual([state], data['state']) self.assertEqual([nonce], data['nonce']) self.assertEqual(['S256'], data['code_challenge_method']) self.assertIn('openid', data['scope'][0]) self.assertIn('code_challenge', data) self.assertTrue(len(verifier) > 0) auth = provider.par_auth_headers[0] self.assertIsNotNone(auth) assert auth is not None self.assertTrue(auth.startswith('Basic ')) @async_test async def test_par_auto_uses_par_only_when_advertised(self) -> None: provider = MockOIDCProvider('par', supports_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, ) client = OIDCClient(config) url, *_ = await client.build_authorization_url(par='auto') query = parse_qs(urlparse(url).query) self.assertIn('request_uri', query) self.assertEqual(1, len(provider.par_requests)) # The default mock provider does not advertise a PAR endpoint. url, *_ = await self.client.build_authorization_url(par='auto') query = parse_qs(urlparse(url).query) self.assertNotIn('request_uri', query) self.assertIn('state', query) @async_test async def test_par_always_without_endpoint_raises(self) -> None: with self.assertRaises(RuntimeError): await self.client.build_authorization_url(par='always') @async_test async def test_required_par_with_never_raises(self) -> None: provider = MockOIDCProvider('req', require_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, ) client = OIDCClient(config) with self.assertRaises(RuntimeError) as ctx: await client.build_authorization_url(par='never') self.assertIn('set par', str(ctx.exception)) self.assertEqual(0, len(provider.par_requests)) @async_test async def test_required_par_with_auto_uses_par(self) -> None: provider = MockOIDCProvider('req', require_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, ) client = OIDCClient(config) url, *_ = await client.build_authorization_url(par='auto') self.assertIn('request_uri', parse_qs(urlparse(url).query)) self.assertEqual(1, len(provider.par_requests)) @async_test async def test_par_error_response_propagates(self) -> None: provider = MockOIDCProvider('par', supports_par=True, par_error=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, par='always', ) client = OIDCClient(config) with self.assertRaises(httpx.HTTPStatusError): await client.build_authorization_url(par='always') self.assertEqual(1, len(provider.par_requests)) @async_test async def test_par_public_client_uses_no_basic_auth(self) -> None: provider = MockOIDCProvider('par', supports_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id='client', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, par='always', ) client = OIDCClient(config) await client.build_authorization_url(par='always') self.assertIsNone(provider.par_auth_headers[0]) self.assertEqual(['client'], provider.par_requests[0]['client_id']) def test_config_par_defaults_to_never_and_validates(self) -> None: config = OIDCConfig( issuer='https://issuer.local', client_id='client', redirect_uri='http://localhost/callback', ) self.assertEqual('never', config.par) with self.assertRaises(ValueError): OIDCConfig( issuer='https://issuer.local', client_id='client', redirect_uri='http://localhost/callback', par='bogus', ) class OIDCParTest(unittest.TestCase): @async_test async def test_login_and_callback_with_par(self) -> None: provider = MockOIDCProvider('par', email='par@example.com', supports_par=True) http_client = httpx.AsyncClient(transport=MockTransport(provider)) store = InMemorySessionStore() session = SessionMixin(store) config = OIDCConfig( issuer=provider.issuer, client_id='client', client_secret='secret', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, par='auto', fetch_userinfo=True, ) oidc = OIDCMixin(config, session=session) app = KayaApp(mixins=[session, oidc]) @app.GET('/profile') @oidc.require_auth async def profile(ctx: HttpContext) -> None: user = oidc.get_user(ctx) await ctx.send_str(200, user.email if user and user.email else 'none') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) self.assertEqual(302, r.status_code) query = parse_qs(urlparse(r.headers['Location']).query) self.assertIn('request_uri', query) self.assertNotIn('state', query) self.assertEqual(1, len(provider.par_requests)) state = provider.par_requests[0]['state'][0] provider.nonce = provider.par_requests[0]['nonce'][0] r = await client.get('/auth/callback', params={'code': 'mock-code', 'state': state}, follow_redirects=False) self.assertEqual(302, r.status_code) r = await client.get('/profile') self.assertEqual('par@example.com', r.text) @async_test async def test_container_applies_policy_per_provider(self) -> None: par_provider = MockOIDCProvider('par', supports_par=True) plain_provider = MockOIDCProvider('plain') par_http = httpx.AsyncClient(transport=MockTransport(par_provider)) plain_http = httpx.AsyncClient(transport=MockTransport(plain_provider)) par_config = OIDCConfig( issuer=par_provider.issuer, client_id='par-client', redirect_uri='http://localhost:8000/auth/withpar/callback', http_client=par_http, provider='withpar', par='always', ) plain_config = OIDCConfig( issuer=plain_provider.issuer, client_id='plain-client', redirect_uri='http://localhost:8000/auth/plain/callback', http_client=plain_http, provider='plain', par='never', ) session = SessionMixin(InMemorySessionStore()) oidc = OIDCProviders([par_config, plain_config], session=session) app = KayaApp(mixins=[session, oidc]) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/withpar/login', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertIn('request_uri', parse_qs(urlparse(r.headers['Location']).query)) r = await client.get('/auth/plain/login', follow_redirects=False) self.assertEqual(302, r.status_code) plain_query = parse_qs(urlparse(r.headers['Location']).query) self.assertNotIn('request_uri', plain_query) self.assertIn('state', plain_query) class OIDCMixinTest(unittest.TestCase): def _build_app(self, fetch_userinfo: bool = False) -> tuple[KayaApp, OIDCMixin, MockOIDCProvider, InMemorySessionStore]: provider = MockOIDCProvider() http_client = httpx.AsyncClient(transport=MockTransport(provider)) store = InMemorySessionStore() session = SessionMixin(store) config = OIDCConfig( issuer=provider.issuer, client_id='client', client_secret='secret', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, fetch_userinfo=fetch_userinfo, ) oidc = OIDCMixin(config, session=session) app = KayaApp(mixins=[session, oidc]) return app, oidc, provider, store def _setup_routes(self, app: KayaApp, oidc: OIDCMixin) -> None: @app.GET('/') async def home(ctx: HttpContext) -> None: await ctx.send_str(200, 'home') @app.GET('/profile') @oidc.require_auth async def profile(ctx: HttpContext) -> None: user = oidc.get_user(ctx) if user is None: await ctx.send_empty(401) return await ctx.send_str(200, f'Hello {user.email}') @app.GET('/refresh') @oidc.require_auth async def refresh(ctx: HttpContext) -> None: new_token = await oidc.refresh_access_token(ctx.session) await ctx.send_str(200, new_token or 'no-token') @app.GET('/id-token') @oidc.require_auth async def id_token(ctx: HttpContext) -> None: stored = ctx.session.get(oidc._key('id_token')) await ctx.send_str(200, stored if isinstance(stored, str) else 'none') async def _login(self, client: httpx.AsyncClient, provider: MockOIDCProvider) -> None: r = await client.get('/auth/login', follow_redirects=False) self.assertEqual(302, r.status_code) query = parse_qs(urlparse(r.headers['Location']).query) provider.nonce = query['nonce'][0] r = await client.get( '/auth/callback', params={'code': 'mock-code', 'state': query['state'][0]}, follow_redirects=False, ) self.assertEqual(302, r.status_code) @async_test async def test_login_redirect(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) self.assertEqual(302, r.status_code) location = r.headers['Location'] self.assertIn(provider.discovery['authorization_endpoint'], location) self.assertIn('Set-Cookie', r.headers) @async_test async def test_callback_success(self) -> None: app, oidc, provider, store = self._build_app(fetch_userinfo=True) self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) self.assertEqual(302, r.status_code) parsed = urlparse(r.headers['Location']) query = parse_qs(parsed.query) state = query['state'][0] nonce = query['nonce'][0] provider.nonce = nonce r = await client.get('/auth/callback', params={'code': 'mock-code', 'state': state}, follow_redirects=False) self.assertEqual(302, r.status_code) self.assertEqual('/', r.headers['Location']) self.assertIn('Set-Cookie', r.headers) r = await client.get('/profile') self.assertEqual(200, r.status_code) self.assertIn('user@example.com', r.text) @async_test async def test_callback_invalid_state(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/callback', params={'code': 'mock-code', 'state': 'wrong'}, follow_redirects=False) self.assertEqual(400, r.status_code) @async_test async def test_callback_invalid_issuer_returns_400(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) query = parse_qs(urlparse(r.headers['Location']).query) provider.nonce = query['nonce'][0] provider.token_issuer = 'https://evil-issuer.local' r = await client.get( '/auth/callback', params={'code': 'mock-code', 'state': query['state'][0]}, follow_redirects=False, ) self.assertEqual(400, r.status_code) @async_test async def test_callback_expired_id_token_returns_400(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) query = parse_qs(urlparse(r.headers['Location']).query) provider.nonce = query['nonce'][0] provider.token_expired = True r = await client.get( '/auth/callback', params={'code': 'mock-code', 'state': query['state'][0]}, follow_redirects=False, ) self.assertEqual(400, r.status_code) @async_test async def test_logout(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/auth/login', follow_redirects=False) parsed = urlparse(r.headers['Location']) query = parse_qs(parsed.query) state = query['state'][0] provider.nonce = query['nonce'][0] await client.get('/auth/callback', params={'code': 'mock-code', 'state': state}, follow_redirects=False) session_id = client.cookies['session_id'] self.assertIn(session_id, store._data) r = await client.get('/auth/logout', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertIn(provider.discovery['end_session_endpoint'], r.headers['Location']) self.assertNotIn(session_id, store._data) @async_test async def test_require_auth_redirect(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/profile', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertEqual('/auth/login', r.headers['Location']) @async_test async def test_refresh_access_token(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, provider) r = await client.get('/refresh') self.assertEqual(200, r.status_code) self.assertEqual('new-access-token', r.text) @async_test async def test_refresh_access_token_rejects_invalid_issuer(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, provider) original_id_token = (await client.get('/id-token')).text provider.token_issuer = 'https://evil-issuer.local' r = await client.get('/refresh') self.assertEqual(200, r.status_code) self.assertEqual('no-token', r.text) # The refreshed ID token was not validated, so nothing was stored. self.assertEqual(original_id_token, (await client.get('/id-token')).text) @async_test async def test_refresh_access_token_rejects_wrong_nonce(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, provider) provider.nonce = 'tampered-nonce' r = await client.get('/refresh') self.assertEqual(200, r.status_code) self.assertEqual('no-token', r.text) @async_test async def test_refresh_access_token_rejects_wrong_subject(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, provider) provider.token_sub = 'other-user' r = await client.get('/refresh') self.assertEqual(200, r.status_code) self.assertEqual('no-token', r.text) @async_test async def test_refresh_access_token_expired_id_token(self) -> None: app, oidc, provider, store = self._build_app() self._setup_routes(app, oidc) transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, provider) provider.token_expired = True r = await client.get('/refresh') self.assertEqual(200, r.status_code) self.assertEqual('no-token', r.text) @async_test async def test_dependency_applied_automatically(self) -> None: # Only pass oidc to KayaApp; SessionMixin should be applied via dependencies. provider = MockOIDCProvider() http_client = httpx.AsyncClient(transport=MockTransport(provider)) store = InMemorySessionStore() session = SessionMixin(store) config = OIDCConfig( issuer=provider.issuer, client_id='client', client_secret='secret', redirect_uri='http://localhost:8000/auth/callback', http_client=http_client, ) oidc = OIDCMixin(config, session=session) app = KayaApp(mixins=[oidc]) @app.GET('/') async def home(ctx: HttpContext) -> None: ctx.session['x'] = 1 await ctx.send_str(200, 'ok') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/') self.assertEqual(200, r.status_code) self.assertIn('Set-Cookie', r.headers) class OIDCMultiProviderTest(unittest.TestCase): def _provider_config(self, provider: MockOIDCProvider, name: str) -> tuple[OIDCConfig, httpx.AsyncClient]: http_client = httpx.AsyncClient(transport=MockTransport(provider)) config = OIDCConfig( issuer=provider.issuer, client_id=f'{name}-client', client_secret='secret', provider=name, redirect_uri=f'http://localhost:8000/auth/{name}/callback', http_client=http_client, fetch_userinfo=True, ) return config, http_client def _build(self, strategy: str = 'multi-identity') -> tuple[KayaApp, OIDCProviders, InMemorySessionStore]: self.google = MockOIDCProvider('google', sub='g-sub', email='g@example.com') self.github = MockOIDCProvider('github', sub='h-sub', email='h@example.com') google_config, self.google_http = self._provider_config(self.google, 'google') github_config, self.github_http = self._provider_config(self.github, 'github') store = InMemorySessionStore() session = SessionMixin(store) oidc = OIDCProviders([google_config, github_config], session=session, strategy=strategy) app = KayaApp(mixins=[session, oidc]) @app.GET('/who') async def who(ctx: HttpContext) -> None: users = oidc.get_users(ctx) await ctx.send_str(200, ','.join(sorted(users)) or 'none') @app.GET('/google-profile') @oidc.require_auth('google') async def google_profile(ctx: HttpContext) -> None: user = oidc.get_user(ctx, 'google') assert user is not None await ctx.send_str(200, user.email or '') @app.GET('/github-profile') @oidc.require_auth('github') async def github_profile(ctx: HttpContext) -> None: user = oidc.get_user(ctx, 'github') assert user is not None await ctx.send_str(200, user.email or '') @app.GET('/refresh-github') @oidc.require_auth('github') async def refresh_github(ctx: HttpContext) -> None: token = await oidc.refresh_access_token(ctx.session, 'github') await ctx.send_str(200, token or 'no-token') return app, oidc, store async def _complete_login(self, client: httpx.AsyncClient, name: str, state: str) -> None: r = await client.get( f'/auth/{name}/callback', params={'code': 'mock-code', 'state': state}, follow_redirects=False, ) self.assertEqual(302, r.status_code) async def _login(self, client: httpx.AsyncClient, provider: MockOIDCProvider, name: str) -> None: r = await client.get(f'/auth/{name}/login', follow_redirects=False) self.assertEqual(302, r.status_code) query = parse_qs(urlparse(r.headers['Location']).query) provider.nonce = query['nonce'][0] await self._complete_login(client, name, query['state'][0]) @async_test async def test_independent_logins(self) -> None: app, oidc, store = self._build() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, self.google, 'google') r = await client.get('/google-profile') self.assertEqual('g@example.com', r.text) # The github provider is untouched by the google login. r = await client.get('/github-profile', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertEqual('/auth/github/login', r.headers['Location']) r = await client.get('/who') self.assertEqual('google', r.text) @async_test async def test_interleaved_login_flows(self) -> None: app, oidc, store = self._build() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: # Start both flows before completing either one. r = await client.get('/auth/google/login', follow_redirects=False) google_query = parse_qs(urlparse(r.headers['Location']).query) r = await client.get('/auth/github/login', follow_redirects=False) github_query = parse_qs(urlparse(r.headers['Location']).query) self.google.nonce = google_query['nonce'][0] self.github.nonce = github_query['nonce'][0] # Completing google must not invalidate the pending github state. await self._complete_login(client, 'google', google_query['state'][0]) await self._complete_login(client, 'github', github_query['state'][0]) r = await client.get('/who') self.assertEqual('github,google', r.text) @async_test async def test_logout_one_provider_keeps_the_other(self) -> None: app, oidc, store = self._build() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, self.google, 'google') await self._login(client, self.github, 'github') session_id = client.cookies['session_id'] self.assertIn(session_id, store._data) r = await client.get('/auth/github/logout', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertIn(self.github.discovery['end_session_endpoint'], r.headers['Location']) # Google is still logged in and the shared session survives. r = await client.get('/google-profile') self.assertEqual('g@example.com', r.text) r = await client.get('/github-profile', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertIn(session_id, store._data) # Logging out the last provider invalidates the session. r = await client.get('/auth/google/logout', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertNotIn(session_id, store._data) @async_test async def test_provider_refresh_is_namespaced(self) -> None: app, oidc, store = self._build() transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, self.google, 'google') await self._login(client, self.github, 'github') # Refreshing github must not clobber google's session state. r = await client.get('/refresh-github') self.assertEqual('new-access-token', r.text) r = await client.get('/google-profile') self.assertEqual('g@example.com', r.text) @async_test async def test_multiple_mixin_instances_compose(self) -> None: google = MockOIDCProvider('google', sub='g-sub', email='g@example.com') github = MockOIDCProvider('github', sub='h-sub', email='h@example.com') google_config, google_http = self._provider_config(google, 'google') github_config, github_http = self._provider_config(github, 'github') store = InMemorySessionStore() session = SessionMixin(store) google_mixin = OIDCMixin( replace(google_config, login_path='/google/login', callback_path='/google/callback', logout_path='/google/logout'), session=session, ) github_mixin = OIDCMixin( replace(github_config, login_path='/github/login', callback_path='/github/callback', logout_path='/github/logout'), session=session, ) app = KayaApp(mixins=[session, google_mixin, github_mixin]) @app.GET('/g') @google_mixin.require_auth async def g(ctx: HttpContext) -> None: user = google_mixin.get_user(ctx) assert user is not None await ctx.send_str(200, user.email or '') @app.GET('/h') @github_mixin.require_auth async def h(ctx: HttpContext) -> None: user = github_mixin.get_user(ctx) assert user is not None await ctx.send_str(200, user.email or '') @app.GET('/who') async def who(ctx: HttpContext) -> None: users = google_mixin.get_users(ctx) providers = google_mixin.logged_in_providers(ctx) summary = ','.join(f'{name}={users[name].email}' for name in sorted(users)) await ctx.send_str(200, f'{",".join(sorted(providers))}|{summary}') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/google/login', follow_redirects=False) query = parse_qs(urlparse(r.headers['Location']).query) google.nonce = query['nonce'][0] r = await client.get('/google/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False) self.assertEqual(302, r.status_code) r = await client.get('/g') self.assertEqual('g@example.com', r.text) r = await client.get('/h', follow_redirects=False) self.assertEqual(302, r.status_code) # Mixin-level helpers see the namespaced user across providers. r = await client.get('/who') self.assertEqual('google|google=g@example.com', r.text) @async_test async def test_single_identity_evicts_previous_provider(self) -> None: app, oidc, store = self._build(strategy='single-identity') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: await self._login(client, self.google, 'google') r = await client.get('/google-profile') self.assertEqual('g@example.com', r.text) # Logging in with github replaces the google identity. await self._login(client, self.github, 'github') r = await client.get('/who') self.assertEqual('github', r.text) r = await client.get('/google-profile', follow_redirects=False) self.assertEqual(302, r.status_code) self.assertEqual('/auth/google/login', r.headers['Location']) r = await client.get('/github-profile') self.assertEqual('h@example.com', r.text) data = store._data[client.cookies['session_id']] self.assertNotIn('oidc:google:user', data) self.assertNotIn('oidc:google:id_token', data) self.assertIn('oidc:github:user', data) @async_test async def test_single_identity_last_completed_login_wins(self) -> None: app, oidc, store = self._build(strategy='single-identity') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: # Start both flows before completing either one; pending flow state # is not evicted, so both callbacks can complete. r = await client.get('/auth/google/login', follow_redirects=False) google_query = parse_qs(urlparse(r.headers['Location']).query) r = await client.get('/auth/github/login', follow_redirects=False) github_query = parse_qs(urlparse(r.headers['Location']).query) self.google.nonce = google_query['nonce'][0] self.github.nonce = github_query['nonce'][0] await self._complete_login(client, 'google', google_query['state'][0]) await self._complete_login(client, 'github', github_query['state'][0]) r = await client.get('/who') self.assertEqual('github', r.text) @async_test async def test_standalone_mixins_default_to_single_identity(self) -> None: google = MockOIDCProvider('google', sub='g-sub', email='g@example.com') github = MockOIDCProvider('github', sub='h-sub', email='h@example.com') google_config, google_http = self._provider_config(google, 'google') github_config, github_http = self._provider_config(github, 'github') session = SessionMixin(InMemorySessionStore()) google_mixin = OIDCMixin( replace(google_config, login_path='/google/login', callback_path='/google/callback', logout_path='/google/logout'), session=session, ) github_mixin = OIDCMixin( replace(github_config, login_path='/github/login', callback_path='/github/callback', logout_path='/github/logout'), session=session, ) app = KayaApp(mixins=[session, google_mixin, github_mixin]) self.assertEqual('single-identity', google_mixin.strategy) @app.GET('/g') @google_mixin.require_auth async def g(ctx: HttpContext) -> None: await ctx.send_str(200, 'google') @app.GET('/h') @github_mixin.require_auth async def h(ctx: HttpContext) -> None: await ctx.send_str(200, 'github') transport = httpx.ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client: r = await client.get('/google/login', follow_redirects=False) query = parse_qs(urlparse(r.headers['Location']).query) google.nonce = query['nonce'][0] r = await client.get('/google/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False) self.assertEqual(302, r.status_code) r = await client.get('/g') self.assertEqual('google', r.text) r = await client.get('/github/login', follow_redirects=False) query = parse_qs(urlparse(r.headers['Location']).query) github.nonce = query['nonce'][0] r = await client.get('/github/callback', params={'code': 'c', 'state': query['state'][0]}, follow_redirects=False) self.assertEqual(302, r.status_code) r = await client.get('/g', follow_redirects=False) self.assertEqual(302, r.status_code) r = await client.get('/h') self.assertEqual('github', r.text) def test_default_strategy_is_single_identity(self) -> None: provider = MockOIDCProvider() config, http = self._provider_config(provider, 'google') session = SessionMixin(InMemorySessionStore()) providers = OIDCProviders([config], session=session) self.assertEqual('single-identity', providers.strategy) mixin = OIDCMixin(config, session=session) self.assertEqual('single-identity', mixin.strategy) def test_invalid_strategy_rejected(self) -> None: provider = MockOIDCProvider() config, http = self._provider_config(provider, 'google') session = SessionMixin(InMemorySessionStore()) with self.assertRaises(ValueError): OIDCMixin(config, session=session, strategy='bogus') with self.assertRaises(ValueError): OIDCProviders([config], session=session, strategy='bogus') def test_config_rejects_invalid_provider(self) -> None: with self.assertRaises(ValueError): OIDCConfig( issuer='https://issuer.local', client_id='client', redirect_uri='http://localhost/callback', provider='bad:name', ) def test_providers_reject_duplicates(self) -> None: config = OIDCConfig( issuer='https://issuer.local', client_id='client', redirect_uri='http://localhost/callback', ) session = SessionMixin(InMemorySessionStore()) with self.assertRaises(ValueError): OIDCProviders([config, config], session=session) def test_providers_require_at_least_one_config(self) -> None: session = SessionMixin(InMemorySessionStore()) with self.assertRaises(ValueError): OIDCProviders([], session=session) def test_providers_reject_bad_base_path(self) -> None: config = OIDCConfig( issuer='https://issuer.local', client_id='client', redirect_uri='http://localhost/callback', ) session = SessionMixin(InMemorySessionStore()) with self.assertRaises(ValueError): OIDCProviders([config], session=session, base_path='auth') if __name__ == '__main__': unittest.main()