Add Pushed Authorization Request (RFC 9126) support

OIDCConfig.par selects never (default), auto, or always. When enabled, build_authorization_url pushes the authorization parameters to the provider's PAR endpoint and redirects the browser with only client_id and request_uri. Providers that require PAR raise a clear error when configured with par='never'. Client authentication follows the token endpoint (HTTP Basic or public client).
2026-10-01 08:23:40 +08:00
1 parent 15dc4b6dad
commit fa8bb0487d
@@ -24,6 +24,9 @@ class MockOIDCProvider:
        issuer: Optional[str] = None,
        sub: str = 'user123',
        email: str = 'user@example.com',
+        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()
@@ -32,6 +35,9 @@ class MockOIDCProvider:
        self.sub = sub
        self.email = email
        self.nonce = f'{name}-nonce-value'
+        self.par_error = par_error
+        self.par_requests: list[dict[str, list[str]]] = []
+        self.par_auth_headers: list[Optional[str]] = []
        self.discovery = {
            'issuer': self.issuer,
            'authorization_endpoint': f'{self.issuer}/auth',
@@ -40,6 +46,10 @@ class MockOIDCProvider:
            '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()
@@ -80,6 +90,20 @@ class MockOIDCProvider:
            return httpx.Response(200, json=self.discovery)
        if path == '/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 {}
@@ -204,6 +228,236 @@ class OIDCClientTest(unittest.TestCase):
        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):