woggioni/kaya
1
0
Fork 0
Code Actions Packages Releases Activity

Add support for multiple OIDC providers

Browse Source 
Namespace per-provider session state as oidc:{provider}:... so several providers can share one session, and add an OIDCProviders mixin that wires multiple configs onto distinct /auth/{provider}/... routes. Logging out of one provider no longer signs the user out of the others.
This commit is contained in:
woggioni
2026-10-01 08:23:40 +08:00
parent 540a7b111e
commit 3af2e7a975
6 changed files with 527 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/kaya-oidc/README.md
+67
View File
@@ -40,6 +40,72 @@ async def profile(ctx: HttpContext):
`OIDCMixin` depends on `SessionMixin`; passing only `oidc` to `KayaApp(mixins=...)`
also works because the app applies mixin dependencies automatically.
## Multiple providers
Use `OIDCProviders` to run several OIDC providers in one application. Each
config's `provider` becomes a route segment under `base_path`
(`/auth/{provider}/login`, `/auth/{provider}/callback` and
`/auth/{provider}/logout`), and all providers share a single `SessionMixin`:
```python
from kaya.oidc import OIDCConfig, OIDCProviders
session = SessionMixin(InMemorySessionStore())
oidc = OIDCProviders(
[
OIDCConfig(
issuer='https://accounts.google.com',
client_id=os.environ['GOOGLE_CLIENT_ID'],
client_secret=os.environ['GOOGLE_CLIENT_SECRET'],
redirect_uri='http://localhost:8000/auth/google/callback',
provider='google',
),
OIDCConfig(
issuer='https://github.com/login/oauth', # or any OIDC issuer
client_id=os.environ['GITHUB_CLIENT_ID'],
client_secret=os.environ['GITHUB_CLIENT_SECRET'],
redirect_uri='http://localhost:8000/auth/github/callback',
provider='github',
),
],
session=session,
base_path='/auth', # optional
)
app = KayaApp(mixins=[session, oidc])
@app.GET('/google-profile')
@oidc.require_auth('google')
async def google_profile(ctx: HttpContext):
user = oidc.get_user(ctx, 'google')
await ctx.send_str(200, f"Hello {user.email or user.sub}")
@app.GET('/who')
async def who(ctx: HttpContext):
users = oidc.get_users(ctx) # {provider: OIDCUser}
providers = oidc.logged_in_providers(ctx) # ['google', ...]
await ctx.send_str(200, ', '.join(users))
```
Each provider's session state is namespaced as `oidc:{provider}:...`, so a user
can be logged into several providers at once. Logging out of one provider only
removes that provider's state and keeps the shared session alive while another
provider is still logged in. Make sure each config's `redirect_uri` points at
that provider's derived callback path.
The composable alternative is to create one `OIDCMixin` per provider, pointing
each at distinct `login_path`/`callback_path`/`logout_path` values and sharing
the same `SessionMixin`:
```python
google = OIDCMixin(google_config, session=session)
github = OIDCMixin(github_config, session=session)
app = KayaApp(mixins=[session, google, github])
```
> **Upgrading:** session keys are now prefixed with `oidc:{provider}:` (the
> default provider is `default`). Existing logged-in users are signed out once
> and simply re-authenticate.
## Features
- Generic OIDC discovery
@@ -50,6 +116,7 @@ also works because the app applies mixin dependencies automatically.
- Optional userinfo endpoint fetch
- Refresh token support
- RP-initiated logout (when provider advertises `end_session_endpoint`)
- Multiple providers in one app via `OIDCProviders` (or several `OIDCMixin` instances)
- Composable with any other `KayaMixin` (RSGI, MCP, etc.)
## Security notes
packages/kaya-oidc/src/kaya/oidc/__init__.py
+2
View File
@@ -1,11 +1,13 @@
from ._client import OIDCClient
from ._config import OIDCConfig
from ._mixin import OIDCMixin, OIDCUser
from ._providers import OIDCProviders
__all__ = [
'OIDCClient',
'OIDCConfig',
'OIDCMixin',
'OIDCProviders',
'OIDCUser',
]
packages/kaya-oidc/src/kaya/oidc/_config.py
+14
View File
@@ -1,13 +1,20 @@
import re
from dataclasses import dataclass, field
from typing import Any, Optional, Sequence
import httpx
_PROVIDER_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$')
@dataclass
class OIDCConfig:
"""Configuration for an OpenID Connect provider.
``provider`` is a stable identifier for this provider. It namespaces the
session state (``oidc:{provider}:...``) so several providers can be used in
the same application without interfering with each other.
``client_secret`` is optional; omit it for public clients.
``fetch_userinfo`` controls whether the userinfo endpoint is queried after
token exchange.
@@ -35,8 +42,15 @@ class OIDCConfig:
'EdDSA',
))
http_client: Optional[httpx.AsyncClient] = None
provider: str = 'default'
def __post_init__(self) -> None:
if not self.provider or _PROVIDER_PATTERN.match(self.provider) is None:
raise ValueError(
'provider must be a non-empty string of letters, digits, '
"underscores or hyphens (got "
f'{self.provider!r})'
)
scopes_list = list(self.scopes)
if 'openid' not in scopes_list:
self.scopes = ('openid', *scopes_list)
packages/kaya-oidc/src/kaya/oidc/_mixin.py
+77 -23
View File
@@ -1,4 +1,4 @@
from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence
from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence, cast
from kaya.core import HttpContext, KayaApp, KayaMixin
from kaya.session import Session, SessionMixin
@@ -10,6 +10,9 @@ from ._config import OIDCConfig
type HttpHandler = Callable[..., Awaitable[None]]
_SESSION_NAMESPACE = 'oidc'
_USER_SUFFIX = 'user'
class OIDCUser(Mapping[str, Any]):
"""Read-only view of the OIDC userinfo / ID token claims."""
@@ -28,7 +31,7 @@ class OIDCUser(Mapping[str, Any]):
@property
def sub(self) -> str:
return self._data['sub']
return cast(str, self._data['sub'])
@property
def email(self) -> Optional[str]:
@@ -56,6 +59,11 @@ class OIDCMixin(KayaMixin):
- ``callback_path`` (default ``/auth/callback``): handles the provider callback.
- ``logout_path`` (default ``/auth/logout``): logs the user out.
Session state is namespaced by ``OIDCConfig.provider`` as
``oidc:{provider}:...`` so several providers can share one session. Use one
``OIDCMixin`` per provider (with distinct route paths), or
:class:`~kaya.oidc.OIDCProviders` to wire several at once.
Example::
session = SessionMixin(InMemorySessionStore())
@@ -73,6 +81,14 @@ class OIDCMixin(KayaMixin):
self._config = config
self._session = session
self._client = OIDCClient(config)
self._prefix = f'{_SESSION_NAMESPACE}:{config.provider}:'
@property
def provider(self) -> str:
return self._config.provider
def _key(self, name: str) -> str:
return f'{self._prefix}{name}'
@property
def dependencies(self) -> Sequence[KayaMixin]:
@@ -83,9 +99,9 @@ class OIDCMixin(KayaMixin):
async def login(ctx: HttpContext) -> None:
auth_url, state, nonce, code_verifier = await self._client.build_authorization_url()
session = self._session_of(ctx)
session['oidc_state'] = state
session['oidc_nonce'] = nonce
session['oidc_code_verifier'] = code_verifier
session[self._key('state')] = state
session[self._key('nonce')] = nonce
session[self._key('code_verifier')] = code_verifier
await ctx.send_empty(302, {'Location': auth_url})
@app.GET(self._config.callback_path)
@@ -108,13 +124,13 @@ class OIDCMixin(KayaMixin):
return
session = self._session_of(ctx)
expected_state = session.get('oidc_state')
expected_state = session.get(self._key('state'))
if state != expected_state:
await ctx.send_str(400, 'Invalid state')
return
nonce = session.get('oidc_nonce', '')
code_verifier = session.get('oidc_code_verifier', '')
nonce = session.get(self._key('nonce'), '')
code_verifier = session.get(self._key('code_verifier'), '')
try:
token_response = await self._client.fetch_token(code, code_verifier)
@@ -131,16 +147,16 @@ class OIDCMixin(KayaMixin):
userinfo = await self._client.fetch_userinfo(access_token)
user.update(userinfo)
session.pop('oidc_state', None)
session.pop('oidc_nonce', None)
session.pop('oidc_code_verifier', None)
session.pop(self._key('state'), None)
session.pop(self._key('nonce'), None)
session.pop(self._key('code_verifier'), None)
session['oidc_user'] = user
session['oidc_id_token'] = id_token
session[self._key('user')] = user
session[self._key('id_token')] = id_token
if isinstance(access_token, str):
session['oidc_access_token'] = access_token
session[self._key('access_token')] = access_token
if 'refresh_token' in token_response:
session['oidc_refresh_token'] = token_response['refresh_token']
session[self._key('refresh_token')] = token_response['refresh_token']
session.regenerate_id()
await ctx.send_empty(302, {'Location': self._config.post_login_redirect})
@@ -150,12 +166,24 @@ class OIDCMixin(KayaMixin):
@app.GET(self._config.logout_path)
async def logout(ctx: HttpContext) -> None:
session = self._session_of(ctx)
id_token = session.get('oidc_id_token')
session.invalidate()
id_token = session.get(self._key('id_token'))
self._logout_session(session)
logout_url = await self._client.build_logout_url(id_token if isinstance(id_token, str) else None)
location = logout_url if logout_url is not None else self._config.post_logout_redirect
await ctx.send_empty(302, {'Location': location})
def _logout_session(self, session: Session) -> None:
"""Drop this provider's session state.
The shared session is only invalidated when no other provider remains
logged in, so logging out of one provider does not sign the user out of
the others.
"""
for key in [key for key in session if key.startswith(self._prefix)]:
del session[key]
if not any(_is_user_key(key) for key in session):
session.invalidate()
@staticmethod
def _session_of(ctx: HttpContext) -> Session:
session = ctx.session
@@ -169,14 +197,31 @@ class OIDCMixin(KayaMixin):
return None
def is_authenticated(self, ctx: HttpContext) -> bool:
return 'oidc_user' in self._session_of(ctx)
return self._key('user') in self._session_of(ctx)
def get_user(self, ctx: HttpContext) -> Optional[OIDCUser]:
user = self._session_of(ctx).get('oidc_user')
user = self._session_of(ctx).get(self._key('user'))
if user is None or not isinstance(user, Mapping):
return None
return OIDCUser(user)
def logged_in_providers(self, ctx: HttpContext) -> list[str]:
"""Return the names of every provider with a user in this session."""
return [_provider_of_user_key(key)
for key in self._session_of(ctx) if _is_user_key(key)]
def get_users(self, ctx: HttpContext) -> dict[str, OIDCUser]:
"""Return the users of every provider logged into this session."""
session = self._session_of(ctx)
users: dict[str, OIDCUser] = {}
for key in session:
if not _is_user_key(key):
continue
data = session.get(key)
if isinstance(data, Mapping):
users[_provider_of_user_key(key)] = OIDCUser(data)
return users
def require_auth(self, handler: HttpHandler) -> HttpHandler:
async def wrapper(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
if not self.is_authenticated(ctx):
@@ -186,18 +231,27 @@ class OIDCMixin(KayaMixin):
return wrapper
async def refresh_access_token(self, session: Session) -> Optional[str]:
refresh_token = session.get('oidc_refresh_token')
refresh_token = session.get(self._key('refresh_token'))
if not isinstance(refresh_token, str):
return None
try:
token_response = await self._client.refresh_token(refresh_token)
access_token = token_response.get('access_token')
if isinstance(access_token, str):
session['oidc_access_token'] = access_token
session[self._key('access_token')] = access_token
if 'id_token' in token_response:
session['oidc_id_token'] = token_response['id_token']
session[self._key('id_token')] = token_response['id_token']
if 'refresh_token' in token_response:
session['oidc_refresh_token'] = token_response['refresh_token']
session[self._key('refresh_token')] = token_response['refresh_token']
return access_token if isinstance(access_token, str) else None
except Exception:
return None
def _is_user_key(key: str) -> bool:
return key.startswith(f'{_SESSION_NAMESPACE}:') and key.endswith(f':{_USER_SUFFIX}')
def _provider_of_user_key(key: str) -> str:
return key[len(_SESSION_NAMESPACE) + 1:-(len(_USER_SUFFIX) + 1)]
packages/kaya-oidc/src/kaya/oidc/_providers.py
+110
View File
@@ -0,0 +1,110 @@
from dataclasses import replace
from types import MappingProxyType
from typing import Callable, Mapping, Optional, Sequence
from kaya.core import HttpContext, KayaApp, KayaMixin
from kaya.session import Session, SessionMixin
from ._config import OIDCConfig
from ._mixin import HttpHandler, OIDCMixin, OIDCUser
class OIDCProviders(KayaMixin):
"""Wire several OpenID Connect providers into one application.
Each config's ``provider`` becomes a route segment under ``base_path``:
``{base_path}/{provider}/login``, ``{base_path}/{provider}/callback`` and
``{base_path}/{provider}/logout``. All providers share the supplied
:class:`~kaya.session.SessionMixin` and namespace their session state by
provider, so one provider's login or logout never disturbs another's.
Example::
session = SessionMixin(InMemorySessionStore())
oidc = OIDCProviders([google_config, github_config], session=session)
app = KayaApp(mixins=[session, oidc])
@app.GET('/profile')
@oidc.require_auth('google')
async def profile(ctx: HttpContext):
user = oidc.get_user(ctx, 'google')
...
"""
def __init__(
self,
configs: Sequence[OIDCConfig],
session: SessionMixin,
base_path: str = '/auth',
) -> None:
if not configs:
raise ValueError('At least one OIDCConfig is required')
if not base_path.startswith('/'):
raise ValueError('base_path must start with "/"')
self._session = session
self._base_path = base_path.rstrip('/')
self._providers: dict[str, OIDCMixin] = {}
for config in configs:
name = config.provider
if name in self._providers:
raise ValueError(f'Duplicate OIDC provider {name!r}')
self._providers[name] = OIDCMixin(
replace(
config,
login_path=f'{self._base_path}/{name}/login',
callback_path=f'{self._base_path}/{name}/callback',
logout_path=f'{self._base_path}/{name}/logout',
),
session=session,
)
@property
def base_path(self) -> str:
return self._base_path
@property
def providers(self) -> Mapping[str, OIDCMixin]:
return MappingProxyType(self._providers)
def provider(self, name: str) -> OIDCMixin:
try:
return self._providers[name]
except KeyError:
raise KeyError(f'Unknown OIDC provider {name!r}') from None
def __getitem__(self, name: str) -> OIDCMixin:
return self.provider(name)
@property
def dependencies(self) -> Sequence[KayaMixin]:
return [self._session]
def apply(self, app: KayaApp) -> None:
for mixin in self._providers.values():
mixin.apply(app)
def is_authenticated(self, ctx: HttpContext, provider: str) -> bool:
return self.provider(provider).is_authenticated(ctx)
def get_user(self, ctx: HttpContext, provider: str) -> Optional[OIDCUser]:
return self.provider(provider).get_user(ctx)
def get_users(self, ctx: HttpContext) -> dict[str, OIDCUser]:
"""Return the user of every provider with a session on ``ctx``."""
users: dict[str, OIDCUser] = {}
for name, mixin in self._providers.items():
user = mixin.get_user(ctx)
if user is not None:
users[name] = user
return users
def logged_in_providers(self, ctx: HttpContext) -> list[str]:
"""Return the names of the providers logged into ``ctx``."""
return [name for name, mixin in self._providers.items() if mixin.is_authenticated(ctx)]
def require_auth(self, provider: str) -> Callable[[HttpHandler], HttpHandler]:
return self.provider(provider).require_auth
async def refresh_access_token(self, session: Session, provider: str) -> Optional[str]:
return await self.provider(provider).refresh_access_token(session)
packages/kaya-oidc/tests/test_oidc.py
+257 -7
View File
@@ -1,7 +1,8 @@
import base64
import unittest
from dataclasses import replace
from time import time
from typing import Mapping
from typing import Mapping, Optional
from urllib.parse import parse_qs, urlparse
import httpx
@@ -11,17 +12,26 @@ from pwo import async_test
from kaya.core import HttpContext, KayaApp
from kaya.session import InMemorySessionStore, SessionMixin
from kaya.oidc import OIDCConfig, OIDCMixin
from kaya.oidc import OIDCConfig, OIDCMixin, OIDCProviders
from kaya.oidc._client import OIDCClient
class MockOIDCProvider:
def __init__(self) -> None:
def __init__(
self,
name: str = 'mock',
*,
issuer: Optional[str] = None,
sub: str = 'user123',
email: str = 'user@example.com',
) -> 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 = 'https://mock-oidc.local'
self.nonce = 'nonce-value'
self.issuer = issuer or f'https://{name}-oidc.local'
self.sub = sub
self.email = email
self.nonce = f'{name}-nonce-value'
self.discovery = {
'issuer': self.issuer,
'authorization_endpoint': f'{self.issuer}/auth',
@@ -48,7 +58,7 @@ class MockOIDCProvider:
now = time()
exp = now - 3600 if expired else now + 3600
payload = {
'sub': 'user123',
'sub': self.sub,
'iss': self.issuer,
'aud': audience,
'iat': now,
@@ -87,7 +97,7 @@ class MockOIDCProvider:
'token_type': 'Bearer',
})
if path == '/userinfo':
return httpx.Response(200, json={'sub': 'user123', 'email': 'user@example.com'})
return httpx.Response(200, json={'sub': self.sub, 'email': self.email})
if path == '/logout':
return httpx.Response(200)
return httpx.Response(404)
@@ -356,5 +366,245 @@ class OIDCMixinTest(unittest.TestCase):
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) -> 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)
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)
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()