kaya/packages/kaya-oidc/tests/test_oidc.py

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()