feat: add OIDC discovery, JWKS, and authorization endpoints

This commit is contained in:
Johan Lundberg 2026-02-16 13:33:40 +01:00
parent 95d184ce0f
commit d8c891af89
No known key found for this signature in database
GPG key ID: A6C152738D03C7D1
4 changed files with 191 additions and 0 deletions

View file

@ -0,0 +1,78 @@
import secrets
from httpx import AsyncClient
def _register_test_client(
client: AsyncClient,
client_id: str = "test-rp",
redirect_uri: str = "http://localhost:9000/callback",
) -> str:
"""Register a test client in the OIDC server. Returns client_secret."""
app = client._transport.app # type: ignore[union-attr]
oidc_server = app.state.oidc_server
client_secret = secrets.token_hex(16)
oidc_server.context.cdb[client_id] = {
"client_id": client_id,
"client_secret": client_secret,
"redirect_uris": [(redirect_uri, {})],
"response_types_supported": ["code"],
"token_endpoint_auth_method": "client_secret_basic",
"scope": ["openid", "profile", "email"],
"allowed_scopes": ["openid", "profile", "email"],
"client_salt": secrets.token_hex(8),
}
oidc_server.keyjar.add_symmetric(client_id, client_secret)
return client_secret
async def test_authorization_redirects_to_login_when_unauthenticated(client: AsyncClient) -> None:
_register_test_client(client)
res = await client.get(
"/authorization",
params={
"response_type": "code",
"client_id": "test-rp",
"redirect_uri": "http://localhost:9000/callback",
"scope": "openid",
"state": "test-state",
"nonce": "test-nonce",
},
follow_redirects=False,
)
assert res.status_code in (302, 303)
assert "/login" in res.headers["location"]
async def test_authorization_stores_auth_request_in_session(client: AsyncClient) -> None:
_register_test_client(client)
res = await client.get(
"/authorization",
params={
"response_type": "code",
"client_id": "test-rp",
"redirect_uri": "http://localhost:9000/callback",
"scope": "openid",
"state": "test-state",
"nonce": "test-nonce",
},
follow_redirects=False,
)
assert res.status_code in (302, 303)
login_res = await client.get("/login")
assert login_res.status_code == 200
async def test_authorization_invalid_client_returns_error(client: AsyncClient) -> None:
res = await client.get(
"/authorization",
params={
"response_type": "code",
"client_id": "nonexistent",
"redirect_uri": "http://evil.com/callback",
"scope": "openid",
"state": "test-state",
},
follow_redirects=False,
)
assert res.status_code >= 400 or "error" in res.text.lower()

View file

@ -0,0 +1,34 @@
from httpx import AsyncClient
async def test_discovery_endpoint_returns_metadata(client: AsyncClient) -> None:
res = await client.get("/.well-known/openid-configuration")
assert res.status_code == 200
data = res.json()
assert data["issuer"] == "http://localhost:8000"
assert "authorization_endpoint" in data
assert "token_endpoint" in data
assert "userinfo_endpoint" in data
assert "jwks_uri" in data
async def test_discovery_response_types_supported(client: AsyncClient) -> None:
res = await client.get("/.well-known/openid-configuration")
data = res.json()
assert "code" in data["response_types_supported"]
async def test_discovery_scopes_supported(client: AsyncClient) -> None:
res = await client.get("/.well-known/openid-configuration")
data = res.json()
assert "openid" in data["scopes_supported"]
async def test_jwks_endpoint_returns_keys(client: AsyncClient) -> None:
res = await client.get("/jwks")
assert res.status_code == 200
data = res.json()
assert "keys" in data
assert len(data["keys"]) > 0
for key in data["keys"]:
assert "kty" in key