Compare commits

15 Commits
Author SHA1 Message Date
woggioni 338c6bd600 Add kaya-openapi package for automatic OpenAPI spec generation
CI / Build Pip package (push) Successful in 2m49s
- New packages/kaya-openapi with OpenAPIMixin, @operation decorator,
  and generate_spec() that walks the routing tree
- Enables kaya-core's Tree.register to expose the original handler
  callback as an instance attribute for metadata introspection
- Registers GET /openapi.json and GET /docs (Swagger UI) routes
- Supports  and  path parameters, docstring
  descriptions, @operation metadata, and excludes wildcard/WS routes
- Adds example/openapi.py, updates CI, README, and requirements
2026-07-25 09:20:19 +00:00
woggioni 1e1e031a56 Fix key-loop in Tree.add to reuse existing children for literal segments after a matcher boundary
CI / Build Pip package (push) Successful in 1m57s
The key-loop unconditionally called self.parse() and overwrote
result.children[key] for literal segments after the walk-down loop
broke at a parameter. This destroyed pre-existing subtrees (with their
method children and handlers) when a second route (e.g. POST) shared
the same literal sub-segments after a parameter.

Added a children.get(key) reuse check before parse(), mirroring the
walk-down loop's child-reuse logic but extended past the boundary
where parameters live in path_matchers, not children.
2026-07-24 12:54:07 +00:00
woggioni 9a314f9bd3 Allow nested routes sharing the same path parameter matcher
CI / Build Pip package (push) Successful in 2m32s
When two routes share the same parameter at the same node position
(e.g. GET /restaurants/${id} and GET /restaurants/${id}/menu),
reuse the existing equivalent matcher instead of raising a conflict.
Non-equivalent matchers (different names, kinds, or glob patterns)
at the same node for the same method still raise ValueError.
2026-07-24 06:46:07 +00:00
woggioni 2dcad41ba1 Rename kaya.session_redis → kaya.session.redis
CI / Build Pip package (push) Successful in 1m56s
- Move src/kaya/session_redis/ → src/kaya/session/redis/
- Update pyproject.toml version_file path
- Update all import references (tests, READMEs, root README)
2026-07-23 16:02:56 +00:00
woggioni c8c8c6052a Rename kaya.session_memcache → kaya.session.memcache
CI / Build Pip package (push) Failing after 1m11s
- Move src/kaya/session_memcache/ → src/kaya/session/memcache/
- Add pkgutil.extend_path to kaya.session for subpackage namespace support
- Update pyproject.toml version_file path
- Update all import references (tests, READMEs, root README)
2026-07-23 15:51:42 +00:00
woggioni dba0c24b1a fix pipeline trigger
CI / Build Pip package (push) Failing after 1m20s
2026-07-23 23:35:15 +08:00
woggioni 2d86a3a187 Add kaya-session-memcache package for memcached-backed session storage 2026-07-23 22:11:06 +08:00
woggioni e469839af3 Add kaya-session-redis package for Redis-backed session storage 2026-07-23 22:11:06 +08:00
woggioni ca023580e4 Fix Daphne ASGI compatibility and document WS handshake cookie limitation
- AsgiContext and AsgiWebSocket now default missing  scope key
  to 'http' / 'ws' respectively (Daphne omits it for websocket scopes)
- Add regression test for websocket scope without scheme
- Update example/session.py WS handler to read the session; cookie must
  be set via HTTP first because common ASGI servers ignore the headers
  field on websocket.accept
- Update README with the same caveat about Granian/Daphne/curl
2026-07-23 22:11:06 +08:00
woggioni e4e00762bb Add websocket session support to kaya-session
- kaya-core: WebSocket ABC gains session attribute and accept(headers=...)
- kaya-core: AsgiWebSocket injects headers into websocket.accept message
- kaya-rsgi: RsgiWebSocket accepts headers param (ignored — Granian's
  accept() takes no args)
- kaya-session: SessionWebSocket wrapper exposes ws.session and injects
  Set-Cookie on accept()
- kaya-session: SessionMixin registers before/after websocket hooks;
  session loaded at connect, persisted on close if modified
- 10 new WV session tests covering read, persist, handshake cookie,
  regenerate, invalidate, isolation
- Example and README updated
2026-07-23 22:11:04 +08:00
woggioni 65f1b79ce8 Fix SessionHttpContext to delegate attributes via __getattr__
The previous implementation eagerly copied ASGI-specific attributes
(pathsend, receive, send) from the wrapped context, which crashed under
RSGI because RsgiContext does not have those attributes. Now
SessionHttpContext delegates all non-overridden attributes to the wrapped
context via __getattr__, making it protocol-agnostic.
2026-07-23 22:09:59 +08:00
woggioni 3ebf079533 Refactor to composable KayaMixin architecture
Replace wrapper-based SessionMiddleware/OIDCApp with KayaMixin subclasses
applied via KayaApp(mixins=[...]). Mixins hook into handle_request and
handle_websocket via before/after hooks, so both ASGI and RSGI keep working.
Mixin dependencies are applied automatically and deduplicated.
2026-07-23 22:09:59 +08:00
woggioni 24a797e3d2 Add kaya-oidc package for OpenID Connect authentication 2026-07-23 22:09:59 +08:00
woggioni 77d1134569 Enforce server-side session idle expiry with sliding TTL 2026-07-23 22:09:59 +08:00
woggioni 97a81a9e41 Add kaya-session package for server-side HTTP session management 2026-07-23 22:09:55 +08:00
57 changed files with 4148 additions and 30 deletions
+65 -5
View File
@@ -2,7 +2,7 @@ name: CI
on:
push:
tags:
- '*'
- 'release/*'
jobs:
build_pip_package:
name: "Build Pip package"
@@ -25,13 +25,33 @@ jobs:
python -m venv .venv
.venv/bin/pip install -r requirements-dev.txt
- name: Check kaya-core
run: |
.venv/bin/python -m mypy -p kaya.rsgi
.venv/bin/python -m unittest discover -s packages/kaya-rsgi/tests
- name: Check kaya-rsgi
run: |
.venv/bin/python -m mypy -p kaya.core
.venv/bin/python -m unittest discover -s packages/kaya-core/tests
- name: Check kaya-rsgi
run: |
.venv/bin/python -m mypy -p kaya.rsgi
.venv/bin/python -m unittest discover -s packages/kaya-rsgi/tests
- name: Check kaya-session
run: |
.venv/bin/python -m mypy -p kaya.session
.venv/bin/python -m unittest discover -s packages/kaya-session/tests
- name: Check kaya-session-memcache
run: |
.venv/bin/python -m mypy -p kaya.session.memcache
.venv/bin/python -m unittest discover -s packages/kaya-session-memcache/tests
- name: Check kaya-session-redis
run: |
.venv/bin/python -m mypy -p kaya.session.redis
.venv/bin/python -m unittest discover -s packages/kaya-session-redis/tests
- name: Check kaya-oidc
run: |
.venv/bin/python -m mypy -p kaya.oidc
.venv/bin/python -m unittest discover -s packages/kaya-oidc/tests
- name: Check kaya-openapi
run: |
.venv/bin/python -m mypy -p kaya.openapi
.venv/bin/python -m unittest discover -s packages/kaya-openapi/tests
- name: Publish kaya-core artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
@@ -48,3 +68,43 @@ jobs:
run: |
.venv/bin/pyproject-build packages/kaya-rsgi
.venv/bin/twine upload --repository gitea packages/kaya-rsgi/dist/*.whl packages/kaya-rsgi/dist/*.tar.gz
- name: Publish kaya-session artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build packages/kaya-session
.venv/bin/twine upload --repository gitea packages/kaya-session/dist/*.whl packages/kaya-session/dist/*.tar.gz
- name: Publish kaya-session-memcache artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build packages/kaya-session-memcache
.venv/bin/twine upload --repository gitea packages/kaya-session-memcache/dist/*.whl packages/kaya-session-memcache/dist/*.tar.gz
- name: Publish kaya-session-redis artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build packages/kaya-session-redis
.venv/bin/twine upload --repository gitea packages/kaya-session-redis/dist/*.whl packages/kaya-session-redis/dist/*.tar.gz
- name: Publish kaya-oidc artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build packages/kaya-oidc
.venv/bin/twine upload --repository gitea packages/kaya-oidc/dist/*.whl packages/kaya-oidc/dist/*.tar.gz
- name: Publish kaya-openapi artifacts
env:
TWINE_REPOSITORY_URL: ${{ vars.PYPI_REGISTRY_URL }}
TWINE_USERNAME: ${{ vars.PUBLISHER_USERNAME }}
TWINE_PASSWORD: ${{ secrets.PUBLISHER_TOKEN }}
run: |
.venv/bin/pyproject-build packages/kaya-openapi
.venv/bin/twine upload --repository gitea packages/kaya-openapi/dist/*.whl packages/kaya-openapi/dist/*.tar.gz
+21 -1
View File
@@ -8,6 +8,11 @@ This repository is a monorepo for the Kaya framework. The code is split into ind
- **kaya-core** — core routing, HTTP/WS abstractions, and ASGI adapter (`packages/kaya-core/`)
- **kaya-rsgi** — RSGI/Granian integration (`packages/kaya-rsgi/`)
- **kaya-session** — server-side HTTP session management (`packages/kaya-session/`)
- **kaya-session-redis** — Redis-backed session storage (`packages/kaya-session-redis/`)
- **kaya-session-memcache** — memcached-backed session storage (`packages/kaya-session-memcache/`)
- **kaya-oidc** — OpenID Connect authentication (`packages/kaya-oidc/`)
- **kaya-openapi** — automatic OpenAPI specification generation (`packages/kaya-openapi/`)
Additional `kaya-*` packages can be added as new directories under `packages/`.
@@ -22,7 +27,7 @@ pip install --index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/si
Install the packages in development mode:
```bash
pip install -e packages/kaya-core -e packages/kaya-rsgi
pip install -e packages/kaya-core -e packages/kaya-rsgi -e packages/kaya-session -e packages/kaya-session-redis -e packages/kaya-session-memcache -e packages/kaya-oidc -e packages/kaya-openapi
```
Run the example:
@@ -36,6 +41,11 @@ python example/hello.py
```bash
python -m unittest discover -s packages/kaya-core/tests
python -m unittest discover -s packages/kaya-rsgi/tests
python -m unittest discover -s packages/kaya-session/tests
python -m unittest discover -s packages/kaya-session-redis/tests
python -m unittest discover -s packages/kaya-session-memcache/tests
python -m unittest discover -s packages/kaya-oidc/tests
python -m unittest discover -s packages/kaya-openapi/tests
```
## Static analysis
@@ -43,6 +53,11 @@ python -m unittest discover -s packages/kaya-rsgi/tests
```bash
mypy -p kaya.core
mypy -p kaya.rsgi
mypy -p kaya.session
mypy -p kaya.session.redis
mypy -p kaya.session.memcache
mypy -p kaya.oidc
mypy -p kaya.openapi
```
## Building packages
@@ -50,4 +65,9 @@ mypy -p kaya.rsgi
```bash
python -m build packages/kaya-core
python -m build packages/kaya-rsgi
python -m build packages/kaya-session
python -m build packages/kaya-session-redis
python -m build packages/kaya-session-memcache
python -m build packages/kaya-oidc
python -m build packages/kaya-openapi
```
+34
View File
@@ -0,0 +1,34 @@
import os
from kaya.core import HttpContext, KayaApp
from kaya.oidc import OIDCConfig, OIDCMixin
from kaya.session import InMemorySessionStore, SessionMixin
session = SessionMixin(InMemorySessionStore())
oidc = OIDCMixin(
OIDCConfig(
issuer=os.environ.get('OIDC_ISSUER', 'https://accounts.google.com'),
client_id=os.environ.get('OIDC_CLIENT_ID', 'replace-me'),
client_secret=os.environ.get('OIDC_CLIENT_SECRET'),
redirect_uri=os.environ.get('OIDC_REDIRECT_URI', 'http://localhost:8000/auth/callback'),
fetch_userinfo=True,
),
session=session,
)
app = KayaApp(mixins=[session, oidc])
@app.GET('/')
async def home(ctx: HttpContext) -> None:
await ctx.send_str(200, 'public 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.name or user.email or user.sub}')
+32
View File
@@ -0,0 +1,32 @@
from kaya.core import HttpContext, KayaApp
from kaya.openapi import OpenAPIMixin, operation
app = KayaApp(mixins=[OpenAPIMixin(
title='Greeting API',
version='1.0.0',
description='Example API documented with kaya-openapi',
)])
@app.GET('/hello')
async def hello(ctx: HttpContext) -> None:
"""Say hello to the world."""
await ctx.send_str(200, 'Hello World')
@app.GET('/hello/${name}')
@operation(summary='Greet someone',
tags=['greetings'],
responses={200: {'description': 'A personalized greeting'}})
async def hello_name(ctx: HttpContext, name: str) -> None:
await ctx.send_str(200, f'Hello {name}')
@app.GET('/square/${x:int}')
@operation(summary='Compute the square of a number', tags=['math'])
async def square(ctx: HttpContext, x: int) -> None:
await ctx.send_str(200, str(x * x))
# serve with an ASGI/RSGI server, e.g.:
# granian --interface rsgi example.openapi:app
# then open http://localhost:8000/docs to browse the API
+45
View File
@@ -0,0 +1,45 @@
from kaya.core import HttpContext, KayaApp, WebSocket
from kaya.session import InMemorySessionStore, SessionMixin
app = KayaApp(mixins=[SessionMixin(InMemorySessionStore())])
@app.GET('/')
async def home(ctx: HttpContext) -> None:
visits = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = visits
await ctx.send_str(200, f'visits: {visits}')
@app.GET('/read')
async def read(ctx: HttpContext) -> None:
visits = ctx.session.get('visits', 0)
await ctx.send_str(200, f'visits: {visits}')
@app.GET('/clear')
async def clear(ctx: HttpContext) -> None:
ctx.session.invalidate()
await ctx.send_str(200, 'session cleared')
@app.websocket('/echo')
async def echo(ws: WebSocket) -> None:
await ws.accept()
async for msg in ws:
if msg.kind == 'text':
await ws.send_text(f"echo: {msg.data}")
elif msg.kind == 'binary':
data = msg.data
assert isinstance(data, bytes)
await ws.send_bytes(data)
@app.websocket('/ws/visits')
async def ws_visits(ws: WebSocket) -> None:
# WebSocket handlers can read the existing session. Most ASGI servers
# (including Granian and Daphne) do not forward the `headers` field of the
# `websocket.accept` message into the HTTP 101 response, so a new session
# cookie cannot be set during the handshake. Use the HTTP `/` endpoint to
# set or refresh the session cookie before connecting here.
await ws.accept()
visits = ws.session.get('visits', 0)
await ws.send_text(f'visits: {visits}')
@@ -1,6 +1,7 @@
from ._app import AbstractKayaApp, KayaApp
from ._http_method import HttpMethod
from ._http_context import HttpContext
from ._mixin import KayaMixin
from ._tree import Tree, PathIterator
from ._path_handler import PathHandler, Matches
from ._websocket import WebSocket, WebSocketMessage
@@ -10,6 +11,7 @@ __all__ = [
'AbstractKayaApp',
'HttpMethod',
'KayaApp',
'KayaMixin',
'HttpContext',
'Tree',
'PathHandler',
+78 -14
View File
@@ -6,9 +6,10 @@ from typing import Callable, Awaitable, Any, Mapping, Sequence, Optional, Tuple,
from pwo import Maybe, AsyncQueueIterator
from ._http_context import HttpContext
from ._http_method import HttpMethod
from ._mixin import KayaMixin
from ._path_handler import Context
from ._types import StrOrStrings
from ._websocket import WebSocket
from ._websocket import WebSocket, WebSocketMessage
from ._asgi import AsgiContext, AsgiWebSocket
from ._tree import Tree
from ._types.asgi import LifespanScope, HTTPScope as ASGIHTTPScope, WebSocketScope as ASGIWebSocketScope
@@ -18,6 +19,10 @@ log = getLogger(__name__)
type HttpHandler = Callable[[HttpContext, Unpack[Any]], Awaitable[None]]
type WebSocketHandler = Callable[[WebSocket, Unpack[Any]], Awaitable[None]]
type BeforeRequestHook = Callable[[HttpContext], Awaitable[Optional[HttpContext]]]
type AfterRequestHook = Callable[[HttpContext], Awaitable[None]]
type BeforeWebSocketHook = Callable[[WebSocket], Awaitable[Optional[WebSocket]]]
type AfterWebSocketHook = Callable[[WebSocket], Awaitable[None]]
class AbstractKayaApp(ABC):
@@ -89,25 +94,84 @@ class AbstractKayaApp(ABC):
class KayaApp(AbstractKayaApp):
_tree: Tree
_mixins: list[KayaMixin]
_applied_ids: set[int]
_before_request_hooks: list[BeforeRequestHook]
_after_request_hooks: list[AfterRequestHook]
_before_websocket_hooks: list[BeforeWebSocketHook]
_after_websocket_hooks: list[AfterWebSocketHook]
def __init__(self) -> None:
def __init__(self, mixins: Sequence[KayaMixin] = ()) -> None:
self._tree = Tree()
self._mixins = []
self._applied_ids = set()
self._before_request_hooks = []
self._after_request_hooks = []
self._before_websocket_hooks = []
self._after_websocket_hooks = []
for mixin in mixins:
self._apply_mixin(mixin)
def _apply_mixin(self, mixin: KayaMixin) -> None:
if id(mixin) in self._applied_ids:
return
for dependency in mixin.dependencies:
self._apply_mixin(dependency)
mixin.apply(self)
self._applied_ids.add(id(mixin))
self._mixins.append(mixin)
def add_before_request_hook(self, hook: BeforeRequestHook) -> None:
self._before_request_hooks.append(hook)
def add_after_request_hook(self, hook: AfterRequestHook) -> None:
self._after_request_hooks.append(hook)
def add_before_websocket_hook(self, hook: BeforeWebSocketHook) -> None:
self._before_websocket_hooks.append(hook)
def add_after_websocket_hook(self, hook: AfterWebSocketHook) -> None:
self._after_websocket_hooks.append(hook)
async def handle_request(self, ctx: HttpContext) -> None:
result = self._tree.get_handler(ctx.path, ctx.method)
if result is not None:
handler, captured = result
await handler.handle_request(ctx, captured)
else:
await ctx.send_empty(404)
for hook in self._before_request_hooks:
new_ctx = await hook(ctx)
if new_ctx is not None:
ctx = new_ctx
try:
result = self._tree.get_handler(ctx.path, ctx.method)
if result is not None:
handler, captured = result
await handler.handle_request(ctx, captured)
else:
await ctx.send_empty(404)
finally:
for hook in reversed(self._after_request_hooks):
await hook(ctx)
async def handle_websocket(self, ws: WebSocket) -> None:
result = self._tree.get_handler(ws.path, HttpMethod.WS)
if result is not None:
handler, captured = result
await handler.handle_request(ws, captured)
else:
await ws.close(1000)
for hook in self._before_websocket_hooks:
new_ws = await hook(ws)
if new_ws is not None:
ws = new_ws
try:
result = self._tree.get_handler(ws.path, HttpMethod.WS)
if result is not None:
handler, captured = result
await handler.handle_request(ws, captured)
else:
await ws.close(1000)
finally:
for hook in reversed(self._after_websocket_hooks):
await hook(ws)
def setup(self, loop: AbstractEventLoop) -> None:
for mixin in self._mixins:
mixin.setup(loop)
def shutdown(self, loop: AbstractEventLoop) -> None:
for mixin in self._mixins:
mixin.shutdown(loop)
def route(self,
paths: StrOrStrings,
+10 -4
View File
@@ -83,11 +83,12 @@ class AsgiContext(HttpContext):
self.path = scope['path']
self.query_string = scope['query_string'].decode()
self.method = HttpMethod(scope['method'])
self.scheme = scope['scheme']
self.scheme = scope.get('scheme', 'http')
self.client = scope['client']
self.server = scope['server']
self.headers = decode_headers(scope['headers'])
self.request_body = request_body_iterator
self.session = (scope.get('state') or {}).get('kaya_session')
async def stream_body(self,
status: int,
@@ -167,13 +168,18 @@ class AsgiWebSocket(WebSocket):
self._send = send
self.path = scope['path']
self.query_string = scope['query_string'].decode()
self.scheme = scope['scheme']
self.scheme = scope.get('scheme', 'ws')
self.client = scope['client']
self.server = scope['server']
self.headers = decode_headers(scope['headers'])
async def accept(self) -> None:
await self._send({'type': 'websocket.accept'})
async def accept(self, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
message: Dict[str, Any] = {'type': 'websocket.accept'}
if headers is not None:
# Emit a list rather than a tuple: wsproto's AcceptConnection (used
# by httpx_ws and others) requires list concatenation.
message['headers'] = list(encode_headers(headers))
await self._send(message)
async def receive(self) -> WebSocketMessage:
message = await self._receive()
@@ -28,6 +28,7 @@ class HttpContext(ABC):
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
request_body: AsyncIterator[bytes]
session: Optional[Any] = None
@abstractmethod
async def stream_body(self,
@@ -0,0 +1,37 @@
from abc import ABC, abstractmethod
from asyncio import AbstractEventLoop
from typing import TYPE_CHECKING, Sequence
if TYPE_CHECKING:
from ._app import KayaApp
class KayaMixin(ABC):
"""Base class for composable Kaya app extensions.
A mixin modifies a ``KayaApp`` instance in place by registering routes,
adding request/websocket hooks, or exposing helper methods on the mixin
instance itself. Because the app remains a ``KayaApp``, both ASGI and RSGI
protocols keep working regardless of which mixins are applied.
Mixins may declare other mixins they depend on via ``dependencies``. The
``KayaApp`` constructor applies dependencies first and guarantees each
mixin is applied at most once.
"""
@property
def dependencies(self) -> Sequence['KayaMixin']:
return ()
@abstractmethod
def apply(self, app: 'KayaApp') -> None:
"""Configure the app: register routes, add hooks, etc."""
pass
def setup(self, loop: AbstractEventLoop) -> None:
"""Called on lifespan startup (default: no-op)."""
pass
def shutdown(self, loop: AbstractEventLoop) -> None:
"""Called on lifespan shutdown (default: no-op)."""
pass
+60 -1
View File
@@ -92,6 +92,16 @@ class Tree:
result = child
key = leaf
while key is not None:
existing = self._find_equivalent_matcher(result, key)
if existing is not None:
result = existing
key = next(it, None)
continue
child = result.children.get(key)
if child is not None:
result = child
key = next(it, None)
continue
new_node = self.parse(key, result)
if isinstance(new_node, Node):
result.children[key] = new_node
@@ -109,6 +119,44 @@ class Tree:
def _supports_method(node: Node | PathMatcher, method: HttpMethod) -> bool:
return None in node.supported_methods or method in node.supported_methods
@staticmethod
def _matcher_identity(leaf: str) -> Optional[Tuple[str, str]]:
start = index_of_with_escape(leaf, '${', '\\', 0)
if start >= 0:
start += 2
end = leaf.index('}', start + 2)
definition = leaf[start:end]
try:
colon = definition.index(':')
except ValueError:
colon = None
if colon is None:
name = definition
kind = 'str'
else:
name = definition[:colon]
kind = definition[colon + 1:]
if kind not in ('str', 'int'):
raise ValueError(f"Unknown kind: '{kind}'")
return (kind, name)
if index_of_with_escape(leaf, '*', '\\', 0) >= 0:
return ('glob', leaf)
return None
def _find_equivalent_matcher(self, node: Node | PathMatcher, leaf: str) -> Optional[PathMatcher]:
identity = self._matcher_identity(leaf)
if identity is None:
return None
kind, key = identity
for existing in node.path_matchers:
if kind == 'str' and isinstance(existing, StrMatcher) and existing.name == key:
return existing
if kind == 'int' and isinstance(existing, IntMatcher) and existing.name == key:
return existing
if kind == 'glob' and isinstance(existing, GlobMatcher) and existing.pattern == key:
return existing
return None
def _check_matcher_conflict(self, node: Node | PathMatcher, method: Optional[HttpMethod]) -> None:
new_is_generic = method is None
for existing in node.path_matchers:
@@ -138,16 +186,27 @@ class Tree:
callback: Callable[[Context, Unpack[Any]], Awaitable[None]],
recursive: bool) -> None:
class Handler(PathHandler):
"""PathHandler created by :meth:`Tree.register`.
The original user callback is exposed through the ``callback``
attribute so that extensions (e.g. ``kaya-openapi``) can inspect
it for metadata such as docstrings or decorator attributes.
"""
callback: Callable[[Context, Unpack[Any]], Awaitable[None]]
async def handle_request(self, ctx: Context, captured: Matches) -> None:
args = Maybe.of_nullable(captured.path).map(lambda it: [it]).or_else([])
await callback(ctx, *args, **captured.kwargs)
await self.callback(ctx, *args, **captured.kwargs)
@property
def recursive(self) -> bool:
return recursive
handler = Handler()
# assigned as an instance attribute (not a class attribute) so that
# the function descriptor protocol does not turn it into a bound method
handler.callback = callback
self.add((p for p in PathIterator(path)), method, handler)
def find_node(self, path: Generator[str, None, None], method: HttpMethod = HttpMethod.GET) \
@@ -1,6 +1,8 @@
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import AsyncIterator, Literal, Mapping, Optional, Sequence, Tuple, Union
from typing import Any, AsyncIterator, Literal, Mapping, Optional, Sequence, Tuple, Union
from ._types.base import StrOrStrings
type WebSocketData = Optional[Union[str, bytes, int]]
@@ -18,9 +20,10 @@ class WebSocket(ABC):
headers: Mapping[str, Sequence[str]]
client: Optional[Tuple[str, int]]
server: Optional[Tuple[str, Optional[int]]]
session: Optional[Any] = None
@abstractmethod
async def accept(self) -> None:
async def accept(self, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
pass
@abstractmethod
+60
View File
@@ -190,3 +190,63 @@ class AsgiTest(unittest.TestCase):
'employee_id': 101325
}, response)
@async_test
async def test_nested_param_routes(self):
app = KayaApp()
@app.GET('/restaurants/${id}')
async def restaurant(ctx: HttpContext, id: str) -> None:
await ctx.send_str(200, f"restaurant:{id}")
@app.GET('/restaurants/${id}/menu')
async def menu(ctx: HttpContext, id: str) -> None:
await ctx.send_str(200, f"menu:{id}")
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("/restaurants/42")
self.assertEqual(200, r.status_code)
self.assertEqual("restaurant:42", r.text)
r = await client.get("/restaurants/42/menu")
self.assertEqual(200, r.status_code)
self.assertEqual("menu:42", r.text)
r = await client.get("/restaurants/42/unknown")
self.assertEqual(404, r.status_code)
@async_test
async def test_nested_param_routes_multiple_methods(self):
app = KayaApp()
@app.GET('/restaurants/${id}')
async def restaurant(ctx: HttpContext, id: str) -> None:
await ctx.send_str(200, f"restaurant:{id}")
@app.GET('/restaurants/${id}/menu')
async def menu_get(ctx: HttpContext, id: str) -> None:
await ctx.send_str(200, f"menu_get:{id}")
@app.POST('/restaurants/${id}/menu')
async def menu_post(ctx: HttpContext, id: str) -> None:
await ctx.send_str(200, f"menu_post:{id}")
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("/restaurants/42")
self.assertEqual(200, r.status_code)
self.assertEqual("restaurant:42", r.text)
r = await client.get("/restaurants/42/menu")
self.assertEqual(200, r.status_code)
self.assertEqual("menu_get:42", r.text)
r = await client.post("/restaurants/42/menu")
self.assertEqual(200, r.status_code)
self.assertEqual("menu_post:42", r.text)
r = await client.put("/restaurants/42/menu")
self.assertEqual(404, r.status_code)
+86 -1
View File
@@ -84,7 +84,14 @@ class TreeTest(unittest.TestCase):
tree = Tree()
tree.add((p for p in ('foo', '*')), None, self.handlers[0])
with self.assertRaises(ValueError):
tree.add((p for p in ('foo', '*')), None, self.handlers[1])
tree.add((p for p in ('foo', '*.md')), None, self.handlers[1])
def test_identical_method_agnostic_matchers_reuse(self):
tree = Tree()
tree.add((p for p in ('foo', '*')), None, self.handlers[0])
tree.add((p for p in ('foo', '*')), None, self.handlers[1])
handler = Maybe.of_nullable(tree.get_handler('/foo/bar', HttpMethod.GET)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], handler)
def test_two_overlapping_method_specific_matchers_raise(self):
tree = Tree()
@@ -101,3 +108,81 @@ class TreeTest(unittest.TestCase):
self.assertIs(self.handlers[0], put_handler)
self.assertIs(self.handlers[1], get_handler)
def test_nested_routes_with_same_param_allowed(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}')), HttpMethod.GET, self.handlers[0])
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.GET, self.handlers[1])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_nested_routes_same_param_reverse_order(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.GET, self.handlers[1])
tree.add((p for p in ('restaurants', '${id}')), HttpMethod.GET, self.handlers[0])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_nested_routes_with_same_int_param_allowed(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id:int}')), HttpMethod.GET, self.handlers[0])
tree.add((p for p in ('restaurants', '${id:int}', 'menu')), HttpMethod.GET, self.handlers[1])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_nested_routes_method_agnostic_reuses_matcher(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}')), HttpMethod.GET, self.handlers[0])
tree.add((p for p in ('restaurants', '${id}', 'menu')), None, self.handlers[1])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.POST)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_different_param_names_still_raise(self):
tree = Tree()
tree.add((p for p in ('a', '${id}')), HttpMethod.GET, self.handlers[0])
with self.assertRaises(ValueError):
tree.add((p for p in ('a', '${name}', 'x')), HttpMethod.GET, self.handlers[1])
def test_different_param_kinds_still_raise(self):
tree = Tree()
tree.add((p for p in ('a', '${id}')), HttpMethod.GET, self.handlers[0])
with self.assertRaises(ValueError):
tree.add((p for p in ('a', '${id:int}', 'x')), HttpMethod.GET, self.handlers[1])
def test_nested_routes_different_methods_share_subtree(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.GET, self.handlers[0])
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.POST, self.handlers[1])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.POST)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_nested_routes_different_methods_reverse_order(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.POST, self.handlers[1])
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.GET, self.handlers[0])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.POST)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
def test_nested_routes_combined_detail_and_menu_methods(self):
tree = Tree()
tree.add((p for p in ('restaurants', '${id}')), HttpMethod.GET, self.handlers[0])
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.GET, self.handlers[1])
tree.add((p for p in ('restaurants', '${id}', 'menu')), HttpMethod.POST, self.handlers[2])
h0 = Maybe.of_nullable(tree.get_handler('/restaurants/42', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h1 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.GET)).map(lambda it: it[0]).or_none()
h2 = Maybe.of_nullable(tree.get_handler('/restaurants/42/menu', HttpMethod.POST)).map(lambda it: it[0]).or_none()
self.assertIs(self.handlers[0], h0)
self.assertIs(self.handlers[1], h1)
self.assertIs(self.handlers[2], h2)
@@ -4,6 +4,7 @@ from pwo import async_test
from httpx_ws import aconnect_ws, WebSocketDisconnect
from httpx_ws.transport import ASGIWebSocketTransport
from kaya.core import KayaApp, WebSocket
from kaya.core._asgi import AsgiWebSocket
class WebSocketTest(unittest.TestCase):
@@ -78,3 +79,75 @@ class WebSocketTest(unittest.TestCase):
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
async with aconnect_ws("/echo", client) as ws:
pass
@async_test
async def test_accept_with_headers(self):
sent_messages = []
async def send(message):
sent_messages.append(message)
async def receive():
return {'type': 'websocket.connect'}
scope = {
'type': 'websocket',
'path': '/echo',
'query_string': b'',
'scheme': 'ws',
'client': ('127.0.0.1', 12345),
'server': ('127.0.0.1', 80),
'headers': [],
}
ws = AsgiWebSocket(scope, receive, send)
await ws.accept(headers={'Set-Cookie': 'sid=abc; Path=/', 'X-Custom': ('a', 'b')})
self.assertEqual(1, len(sent_messages))
message = sent_messages[0]
self.assertEqual('websocket.accept', message['type'])
self.assertIn((b'Set-Cookie', b'sid=abc; Path=/'), message['headers'])
self.assertIn((b'X-Custom', b'a'), message['headers'])
self.assertIn((b'X-Custom', b'b'), message['headers'])
@async_test
async def test_accept_without_headers(self):
sent_messages = []
async def send(message):
sent_messages.append(message)
async def receive():
return {'type': 'websocket.connect'}
scope = {
'type': 'websocket',
'path': '/echo',
'query_string': b'',
'scheme': 'ws',
'client': ('127.0.0.1', 12345),
'server': ('127.0.0.1', 80),
'headers': [],
}
ws = AsgiWebSocket(scope, receive, send)
await ws.accept()
self.assertEqual(1, len(sent_messages))
self.assertEqual({'type': 'websocket.accept'}, sent_messages[0])
@async_test
async def test_websocket_scope_without_scheme(self):
# Daphne omits the optional `scheme` key from websocket scopes.
async def send(message):
pass
async def receive():
return {'type': 'websocket.connect'}
scope = {
'type': 'websocket',
'path': '/echo',
'query_string': b'',
'client': ('127.0.0.1', 12345),
'server': ('127.0.0.1', 80),
'headers': [],
}
ws = AsgiWebSocket(scope, receive, send)
self.assertEqual('ws', ws.scheme)
+73
View File
@@ -0,0 +1,73 @@
# kaya-oidc
OpenID Connect authentication for the Kaya web framework.
Built on top of `kaya-session` and implements the OIDC **Authorization Code
Flow with PKCE**.
## Usage
```python
import os
from kaya.core import HttpContext, KayaApp
from kaya.session import SessionMixin, InMemorySessionStore
from kaya.oidc import OIDCConfig, OIDCMixin
session = SessionMixin(InMemorySessionStore())
oidc = OIDCMixin(
OIDCConfig(
issuer=os.environ['OIDC_ISSUER'],
client_id=os.environ['OIDC_CLIENT_ID'],
client_secret=os.environ.get('OIDC_CLIENT_SECRET'),
redirect_uri='http://localhost:8000/auth/callback',
fetch_userinfo=True,
),
session=session,
)
app = KayaApp(mixins=[session, oidc])
@app.GET('/')
async def home(ctx: HttpContext):
await ctx.send_str(200, 'public home')
@app.GET('/profile')
@oidc.require_auth
async def profile(ctx: HttpContext):
user = oidc.get_user(ctx)
await ctx.send_str(200, f'Hello {user.email or user.sub}')
```
`OIDCMixin` depends on `SessionMixin`; passing only `oidc` to `KayaApp(mixins=...)`
also works because the app applies mixin dependencies automatically.
## Features
- Generic OIDC discovery
- Authorization Code Flow with PKCE (S256)
- ID token signature validation with JWKS
- `state` and `nonce` protection
- Session fixation defense via `regenerate_id()` after login
- Optional userinfo endpoint fetch
- Refresh token support
- RP-initiated logout (when provider advertises `end_session_endpoint`)
- Composable with any other `KayaMixin` (RSGI, MCP, etc.)
## Security notes
- The `none` signing algorithm is rejected by default.
- Only algorithms listed in `OIDCConfig.allowed_id_token_algorithms` are accepted.
- Always use HTTPS for `redirect_uri` in production.
## Supported flows
Only the Authorization Code Flow with PKCE is supported. Implicit and Hybrid
flows are intentionally not implemented.
## Supported algorithms
ID token signature verification supports:
`RS256`, `RS384`, `RS512`, `PS256`, `PS384`, `PS512`,
`ES256`, `ES384`, `ES512`, and `EdDSA`.
HMAC algorithms (`HS*`) are disabled by default and can be enabled by adding
them to `allowed_id_token_algorithms` if your provider uses them.
+59
View File
@@ -0,0 +1,59 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya-oidc"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "OpenID Connect authentication for the Kaya lightweight ASGI web framework"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'Programming Language :: Python :: 3',
]
dependencies = [
"kaya-core",
"kaya-session",
"httpx",
"PyJWT[crypto]",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
root = "../.."
version_file = "src/kaya/oidc/_version.py"
[tool.setuptools_scm.tag]
prefix = "release/"
@@ -0,0 +1,11 @@
from ._client import OIDCClient
from ._config import OIDCConfig
from ._mixin import OIDCMixin, OIDCUser
__all__ = [
'OIDCClient',
'OIDCConfig',
'OIDCMixin',
'OIDCUser',
]
+182
View File
@@ -0,0 +1,182 @@
from typing import Any, Mapping, Optional, Sequence, cast
from urllib.parse import urlencode
import httpx
import jwt
from jwt import PyJWK
from ._config import OIDCConfig
from ._utils import generate_nonce, generate_pkce, generate_state
class OIDCClient:
"""Low-level OIDC client implementing discovery, PKCE, and token validation."""
def __init__(self, config: OIDCConfig) -> None:
self._config = config
self._metadata: Optional[Mapping[str, Any]] = None
self._jwks: Optional[Mapping[str, Any]] = None
async def _ensure_metadata(self) -> Mapping[str, Any]:
if self._metadata is None:
url = f"{self._config.issuer.rstrip('/')}/.well-known/openid-configuration"
resp = await self._request('GET', url)
resp.raise_for_status()
self._metadata = cast(Mapping[str, Any], resp.json())
return self._metadata
async def _ensure_jwks(self) -> Mapping[str, Any]:
if self._jwks is None:
metadata = await self._ensure_metadata()
jwks_uri = metadata.get('jwks_uri')
if not isinstance(jwks_uri, str):
raise RuntimeError('OIDC provider does not advertise a jwks_uri')
resp = await self._request('GET', jwks_uri)
resp.raise_for_status()
self._jwks = cast(Mapping[str, Any], resp.json())
return self._jwks
async def _request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
if self._config.http_client is not None:
return await self._config.http_client.request(method, url, **kwargs)
async with httpx.AsyncClient() as client:
return await client.request(method, url, **kwargs)
async def build_authorization_url(self) -> tuple[str, str, str, str]:
"""Return (authorization_url, state, nonce, code_verifier)."""
metadata = await self._ensure_metadata()
authorization_endpoint = metadata.get('authorization_endpoint')
if not isinstance(authorization_endpoint, str):
raise RuntimeError('OIDC provider does not advertise an authorization_endpoint')
state = generate_state()
nonce = generate_nonce()
code_verifier, code_challenge = generate_pkce()
params = {
'client_id': self._config.client_id,
'response_type': 'code',
'scope': ' '.join(self._config.scopes),
'redirect_uri': self._config.redirect_uri,
'state': state,
'nonce': nonce,
'code_challenge': code_challenge,
'code_challenge_method': 'S256',
}
url = authorization_endpoint + '?' + urlencode(params)
return url, state, nonce, code_verifier
async def fetch_token(self, code: str, code_verifier: str) -> Mapping[str, Any]:
metadata = await self._ensure_metadata()
token_endpoint = metadata.get('token_endpoint')
if not isinstance(token_endpoint, str):
raise RuntimeError('OIDC provider does not advertise a token_endpoint')
data = {
'grant_type': 'authorization_code',
'code': code,
'redirect_uri': self._config.redirect_uri,
'client_id': self._config.client_id,
'code_verifier': code_verifier,
}
auth: Optional[tuple[str, str]] = None
if self._config.client_secret is not None:
auth = (self._config.client_id, self._config.client_secret)
resp = await self._request('POST', token_endpoint, data=data, auth=auth)
resp.raise_for_status()
return cast(Mapping[str, Any], resp.json())
async def refresh_token(self, refresh_token: str) -> Mapping[str, Any]:
metadata = await self._ensure_metadata()
token_endpoint = metadata.get('token_endpoint')
if not isinstance(token_endpoint, str):
raise RuntimeError('OIDC provider does not advertise a token_endpoint')
data = {
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'client_id': self._config.client_id,
}
auth: Optional[tuple[str, str]] = None
if self._config.client_secret is not None:
auth = (self._config.client_id, self._config.client_secret)
resp = await self._request('POST', token_endpoint, data=data, auth=auth)
resp.raise_for_status()
return cast(Mapping[str, Any], resp.json())
async def validate_id_token(self, id_token: str, nonce: str) -> Mapping[str, Any]:
header = jwt.get_unverified_header(id_token)
alg = header.get('alg')
if not isinstance(alg, str):
raise ValueError('ID token header is missing alg')
if alg == 'none':
if not self._config.allow_unsigned_id_tokens:
raise ValueError('Unsigned ID tokens are not allowed')
payload = jwt.decode(
id_token,
options={'verify_signature': False},
)
else:
if alg not in self._config.allowed_id_token_algorithms:
raise ValueError(f'ID token algorithm {alg} is not allowed')
jwks = await self._ensure_jwks()
signing_key = self._get_signing_key(jwks, header.get('kid'), alg)
payload = jwt.decode(
id_token,
signing_key.key,
algorithms=[alg],
issuer=self._config.issuer,
audience=self._config.client_id,
)
if not isinstance(payload, dict):
raise ValueError('ID token payload is not a JSON object')
if payload.get('nonce') != nonce:
raise ValueError('ID token nonce mismatch')
return payload
def _get_signing_key(self, jwks: Mapping[str, Any], kid: Optional[str], alg: str) -> PyJWK:
keys = jwks.get('keys')
if not isinstance(keys, Sequence):
raise ValueError('JWKS contains no keys')
if kid is not None:
for key in keys:
if isinstance(key, Mapping) and key.get('kid') == kid:
return PyJWK(cast(dict[str, Any], dict(key)), algorithm=alg)
raise ValueError(f'No signing key found for kid {kid}')
if len(keys) == 1 and isinstance(keys[0], Mapping):
return PyJWK(cast(dict[str, Any], dict(keys[0])), algorithm=alg)
raise ValueError('ID token has no kid and JWKS contains multiple keys')
async def fetch_userinfo(self, access_token: str) -> Mapping[str, Any]:
metadata = await self._ensure_metadata()
userinfo_endpoint = metadata.get('userinfo_endpoint')
if not isinstance(userinfo_endpoint, str):
raise RuntimeError('OIDC provider does not advertise a userinfo_endpoint')
resp = await self._request(
'GET',
userinfo_endpoint,
headers={'Authorization': f'Bearer {access_token}'},
)
resp.raise_for_status()
return cast(Mapping[str, Any], resp.json())
async def build_logout_url(self, id_token: Optional[str] = None) -> Optional[str]:
metadata = await self._ensure_metadata()
end_session_endpoint = metadata.get('end_session_endpoint')
if not isinstance(end_session_endpoint, str):
return None
params: dict[str, str] = {
'post_logout_redirect_uri': self._config.post_logout_redirect,
}
if id_token is not None:
params['id_token_hint'] = id_token
return end_session_endpoint + '?' + urlencode(params)
@@ -0,0 +1,42 @@
from dataclasses import dataclass, field
from typing import Any, Optional, Sequence
import httpx
@dataclass
class OIDCConfig:
"""Configuration for an OpenID Connect provider.
``client_secret`` is optional; omit it for public clients.
``fetch_userinfo`` controls whether the userinfo endpoint is queried after
token exchange.
``allowed_id_token_algorithms`` lists the JWS algorithms the client will
accept when validating the ID token. The ``none`` algorithm is rejected
unless ``allow_unsigned_id_tokens`` is set to ``True``.
"""
issuer: str
client_id: str
redirect_uri: str
client_secret: Optional[str] = None
scopes: Sequence[str] = ('openid', 'email', 'profile')
login_path: str = '/auth/login'
callback_path: str = '/auth/callback'
logout_path: str = '/auth/logout'
post_login_redirect: str = '/'
post_logout_redirect: str = '/'
fetch_userinfo: bool = False
allow_unsigned_id_tokens: bool = False
allowed_id_token_algorithms: Sequence[str] = field(default_factory=lambda: (
'RS256', 'RS384', 'RS512',
'PS256', 'PS384', 'PS512',
'ES256', 'ES384', 'ES512',
'EdDSA',
))
http_client: Optional[httpx.AsyncClient] = None
def __post_init__(self) -> None:
scopes_list = list(self.scopes)
if 'openid' not in scopes_list:
self.scopes = ('openid', *scopes_list)
+203
View File
@@ -0,0 +1,203 @@
from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence
from kaya.core import HttpContext, KayaApp, KayaMixin
from kaya.session import Session, SessionMixin
from urllib.parse import parse_qs
from ._client import OIDCClient
from ._config import OIDCConfig
type HttpHandler = Callable[..., Awaitable[None]]
class OIDCUser(Mapping[str, Any]):
"""Read-only view of the OIDC userinfo / ID token claims."""
def __init__(self, data: Mapping[str, Any]) -> None:
self._data = dict(data)
def __getitem__(self, key: str) -> Any:
return self._data[key]
def __iter__(self) -> Any:
return iter(self._data)
def __len__(self) -> int:
return len(self._data)
@property
def sub(self) -> str:
return self._data['sub']
@property
def email(self) -> Optional[str]:
return self._data.get('email')
@property
def name(self) -> Optional[str]:
return self._data.get('name')
@property
def picture(self) -> Optional[str]:
return self._data.get('picture')
class OIDCMixin(KayaMixin):
"""Kaya mixin adding OpenID Connect authentication.
Depends on :class:`~kaya.session.SessionMixin` so that OIDC state, nonce,
and user data can be stored in ``ctx.session``. The dependency is applied
automatically by ``KayaApp``.
Registers three routes on the app:
- ``login_path`` (default ``/auth/login``): redirects to the OIDC provider.
- ``callback_path`` (default ``/auth/callback``): handles the provider callback.
- ``logout_path`` (default ``/auth/logout``): logs the user out.
Example::
session = SessionMixin(InMemorySessionStore())
oidc = OIDCMixin(config, session=session)
app = KayaApp(mixins=[session, oidc])
@app.GET('/profile')
@oidc.require_auth
async def profile(ctx: HttpContext):
user = oidc.get_user(ctx)
...
"""
def __init__(self, config: OIDCConfig, session: SessionMixin) -> None:
self._config = config
self._session = session
self._client = OIDCClient(config)
@property
def dependencies(self) -> Sequence[KayaMixin]:
return [self._session]
def apply(self, app: KayaApp) -> None:
@app.GET(self._config.login_path)
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
await ctx.send_empty(302, {'Location': auth_url})
@app.GET(self._config.callback_path)
async def callback(ctx: HttpContext) -> None:
query = parse_qs(ctx.query_string)
code = self._first_value(query.get('code'))
state = self._first_value(query.get('state'))
error = self._first_value(query.get('error'))
error_description = self._first_value(query.get('error_description'))
if error is not None:
message = f'OIDC error: {error}'
if error_description is not None:
message += f': {error_description}'
await ctx.send_str(400, message)
return
if code is None or state is None:
await ctx.send_str(400, 'Missing code or state')
return
session = self._session_of(ctx)
expected_state = session.get('oidc_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', '')
try:
token_response = await self._client.fetch_token(code, code_verifier)
id_token = token_response.get('id_token')
if not isinstance(id_token, str):
await ctx.send_str(400, 'No id_token in token response')
return
claims = await self._client.validate_id_token(id_token, nonce)
user: dict[str, Any] = dict(claims)
access_token = token_response.get('access_token')
if self._config.fetch_userinfo and isinstance(access_token, str):
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['oidc_user'] = user
session['oidc_id_token'] = id_token
if isinstance(access_token, str):
session['oidc_access_token'] = access_token
if 'refresh_token' in token_response:
session['oidc_refresh_token'] = token_response['refresh_token']
session.regenerate_id()
await ctx.send_empty(302, {'Location': self._config.post_login_redirect})
except ValueError as exc:
await ctx.send_str(400, f'Authentication failed: {exc}')
@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()
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})
@staticmethod
def _session_of(ctx: HttpContext) -> Session:
session = ctx.session
assert isinstance(session, Session)
return session
@staticmethod
def _first_value(values: Optional[Sequence[str]]) -> Optional[str]:
if values and len(values) > 0:
return values[0]
return None
def is_authenticated(self, ctx: HttpContext) -> bool:
return 'oidc_user' in self._session_of(ctx)
def get_user(self, ctx: HttpContext) -> Optional[OIDCUser]:
user = self._session_of(ctx).get('oidc_user')
if user is None or not isinstance(user, Mapping):
return None
return OIDCUser(user)
def require_auth(self, handler: HttpHandler) -> HttpHandler:
async def wrapper(ctx: HttpContext, *args: Any, **kwargs: Any) -> None:
if not self.is_authenticated(ctx):
await ctx.send_empty(302, {'Location': self._config.login_path})
return
await handler(ctx, *args, **kwargs)
return wrapper
async def refresh_access_token(self, session: Session) -> Optional[str]:
refresh_token = session.get('oidc_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
if 'id_token' in token_response:
session['oidc_id_token'] = token_response['id_token']
if 'refresh_token' in token_response:
session['oidc_refresh_token'] = token_response['refresh_token']
return access_token if isinstance(access_token, str) else None
except Exception:
return None
@@ -0,0 +1,22 @@
import base64
import hashlib
import secrets
def generate_state() -> str:
"""Generate a random CSRF state parameter."""
return secrets.token_urlsafe(32)
def generate_nonce() -> str:
"""Generate a random nonce for replay protection."""
return secrets.token_urlsafe(32)
def generate_pkce() -> tuple[str, str]:
"""Return a PKCE (code_verifier, code_challenge) pair using S256."""
verifier = secrets.token_urlsafe(64)
challenge = base64.urlsafe_b64encode(
hashlib.sha256(verifier.encode()).digest()
).rstrip(b'=').decode()
return verifier, challenge
+360
View File
@@ -0,0 +1,360 @@
import base64
import unittest
from time import time
from typing import Mapping
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
from kaya.oidc._client import OIDCClient
class MockOIDCProvider:
def __init__(self) -> 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.discovery = {
'issuer': self.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',
}
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: str, audience: str, expired: bool = False, wrong_nonce: bool = False) -> str:
now = time()
exp = now - 3600 if expired else now + 3600
payload = {
'sub': 'user123',
'iss': self.issuer,
'aud': audience,
'iat': now,
'exp': exp,
'nonce': 'wrong-nonce' if wrong_nonce else nonce,
}
return jwt.encode(
payload,
self.private_key,
algorithm='RS256',
headers={'kid': self.kid},
)
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 == '/.well-known/openid-configuration':
return httpx.Response(200, json=self.discovery)
if path == '/jwks':
return httpx.Response(200, json={'keys': [self._public_key_to_jwk()]})
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_id_token('refreshed-nonce', 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_id_token(self.nonce, data.get('client_id', 'client')),
'token_type': 'Bearer',
})
if path == '/userinfo':
return httpx.Response(200, json={'sub': 'user123', 'email': 'user@example.com'})
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:
token = self.provider.issue_id_token('nonce-value', 'client')
config = OIDCConfig(
issuer='https://wrong-issuer.local',
client_id='client',
redirect_uri='http://localhost:8000/auth/callback',
http_client=self.http_client,
)
client = OIDCClient(config)
with self.assertRaises(jwt.InvalidIssuerError):
await client.validate_id_token(token, 'nonce-value')
@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_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'])
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')
@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_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:
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)
r = await client.get('/refresh')
self.assertEqual(200, r.status_code)
self.assertEqual('new-access-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)
if __name__ == '__main__':
unittest.main()
+93
View File
@@ -0,0 +1,93 @@
# kaya-openapi
Automatic [OpenAPI](https://www.openapis.org/) specification generation for the
[Kaya](https://github.com/woggioni/kaya) lightweight ASGI web framework.
The package provides an `OpenAPIMixin` that inspects a `KayaApp`'s routing
tree and serves:
- an OpenAPI 3.1 JSON document (default: `GET /openapi.json`)
- a Swagger UI page to browse it interactively (default: `GET /docs`)
## Usage
```python
from kaya.core import HttpContext, KayaApp
from kaya.openapi import OpenAPIMixin, operation
app = KayaApp(mixins=[OpenAPIMixin(title='My API', version='1.0.0')])
@app.GET('/users/${user_id:int}')
@operation(summary='Get a user',
tags=['users'],
responses={
200: {'description': 'The user'},
404: {'description': 'User not found'},
})
async def get_user(ctx: HttpContext, user_id: int) -> None:
...
```
Run the app with any ASGI/RSGI server and open `http://localhost:8000/docs`.
## How routes are mapped
- Static segments and parameters are converted to OpenAPI path templating:
- `/users/${user_id}` → `/users/{user_id}` (string path parameter)
- `/users/${user_id:int}` → `/users/{user_id}` (integer path parameter)
- Wildcard routes (`*`) are **skipped**: they cannot be expressed in OpenAPI
path syntax.
- Websocket routes are **skipped**: OpenAPI does not model websockets.
- Method-agnostic routes (registered with `app.route(path)` without methods)
are documented under **all** standard HTTP methods, since they respond to
all of them.
- The mixin's own endpoints are excluded from the document unless
`include_self=True`.
The document is generated on every request to the spec endpoint, so routes
registered after the mixin is applied are always included.
## Operation metadata
The `@operation` decorator attaches OpenAPI metadata to a route handler.
All fragments are plain dicts merged verbatim into the generated operation
object, so any valid OpenAPI 3.1 construct can be used:
```python
@operation(summary='...', # operation summary
description='...', # defaults to the handler docstring
tags=['users'],
operation_id='getUser',
request_body={...}, # OpenAPI requestBody object
responses={200: {...}}, # per-status-code response objects
parameters=[...], # extra/overriding parameter objects
deprecated=False,
hidden=False) # exclude from the document
```
`parameters` entries whose `name` and `in` match an auto-generated path
parameter override it; all others are appended.
## Configuration
```python
OpenAPIMixin(
title='My API', # info.title (required)
version='1.0.0', # info.version (required)
description='', # info.description
spec_path='/openapi.json', # where the JSON document is served
docs_path='/docs', # where Swagger UI is served
servers=[{'url': 'https://api.example.com'}],
openapi_version='3.1.0',
include_self=False, # include spec/docs endpoints in the document
)
```
The document can also be generated programmatically without serving it:
```python
from kaya.openapi import generate_spec
spec = generate_spec(app, title='My API', version='1.0.0')
```
+56
View File
@@ -0,0 +1,56 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya-openapi"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "Automatic OpenAPI specification generation for the Kaya lightweight ASGI web framework"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'Programming Language :: Python :: 3',
]
dependencies = [
"kaya-core",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
root = "../.."
version_file = "src/kaya/openapi/_version.py"
[tool.setuptools_scm.tag]
prefix = "release/"
@@ -0,0 +1,10 @@
from ._metadata import operation
from ._mixin import OpenAPIMixin
from ._spec import generate_spec
__all__ = [
'OpenAPIMixin',
'generate_spec',
'operation',
]
@@ -0,0 +1,84 @@
from typing import Any, Awaitable, Callable, Mapping, Optional, Sequence, TypeVar
F = TypeVar('F', bound=Callable[..., Awaitable[None]])
METADATA_ATTR = '__kaya_openapi__'
#: Metadata attached to route handler functions by :func:`operation`.
#: Values are raw OpenAPI fragments (plain dicts) merged verbatim into the
#: generated operation object.
type OperationMetadata = Mapping[str, Any]
def operation(summary: Optional[str] = None,
description: Optional[str] = None,
tags: Optional[Sequence[str]] = None,
request_body: Optional[Mapping[str, Any]] = None,
responses: Optional[Mapping[int | str, Mapping[str, Any]]] = None,
parameters: Optional[Sequence[Mapping[str, Any]]] = None,
operation_id: Optional[str] = None,
deprecated: bool = False,
hidden: bool = False) -> Callable[[F], F]:
"""Attach OpenAPI metadata to a Kaya route handler.
The metadata is stored on the function itself and picked up by
:class:`~kaya.openapi.OpenAPIMixin` when generating the specification.
All schema fragments are plain dicts inserted verbatim into the generated
OpenAPI document, so any valid OpenAPI 3.1 construct can be used.
Example::
@app.GET('/users/${user_id:int}')
@operation(summary='Get a user',
tags=['users'],
responses={200: {'description': 'The user'},
404: {'description': 'User not found'}})
async def get_user(ctx: HttpContext, user_id: int) -> None:
...
:param summary: short summary of the operation
:param description: longer description (defaults to the handler docstring)
:param tags: list of OpenAPI tags
:param request_body: OpenAPI ``requestBody`` object
:param responses: mapping of status code (or ``'default'``) to OpenAPI
response objects
:param parameters: extra OpenAPI parameter objects merged with the
auto-generated path parameters (entries whose ``name`` matches a path
parameter override the auto-generated one)
:param operation_id: explicit OpenAPI ``operationId``
:param deprecated: mark the operation as deprecated
:param hidden: exclude the operation from the generated specification
"""
def decorator(func: F) -> F:
metadata: dict[str, Any] = {}
if summary is not None:
metadata['summary'] = summary
if description is not None:
metadata['description'] = description
if tags is not None:
metadata['tags'] = list(tags)
if request_body is not None:
metadata['request_body'] = dict(request_body)
if responses is not None:
metadata['responses'] = {str(k): dict(v) for k, v in responses.items()}
if parameters is not None:
metadata['parameters'] = [dict(p) for p in parameters]
if operation_id is not None:
metadata['operation_id'] = operation_id
if deprecated:
metadata['deprecated'] = True
if hidden:
metadata['hidden'] = True
setattr(func, METADATA_ATTR, metadata)
return func
return decorator
def get_metadata(handler: Any) -> OperationMetadata:
"""Return the metadata attached by :func:`operation`, or an empty mapping."""
metadata = getattr(handler, METADATA_ATTR, None)
if isinstance(metadata, Mapping):
return metadata
return {}
@@ -0,0 +1,110 @@
import json
from html import escape
from typing import Any, Mapping, Optional, Sequence
from kaya.core import HttpContext, KayaApp, KayaMixin
from ._spec import generate_spec
_DOCS_PAGE = """<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>{title} - API documentation</title>
<link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css" />
</head>
<body>
<div id="swagger-ui"></div>
<script src="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js" crossorigin></script>
<script>
window.onload = () => {{
window.ui = SwaggerUIBundle({{
url: '{spec_path}',
dom_id: '#swagger-ui',
}});
}};
</script>
</body>
</html>
"""
class OpenAPIMixin(KayaMixin):
"""Kaya mixin serving an auto-generated OpenAPI specification.
Registers two routes on the app:
- ``spec_path`` (default ``/openapi.json``): the OpenAPI document,
generated on each request from the application's routing tree so that
routes registered after the mixin are always included.
- ``docs_path`` (default ``/docs``): a Swagger UI page rendering the
specification (assets are loaded from a CDN).
Routes are converted as follows:
- ``${name}`` path segments become ``{name}`` string path parameters,
``${name:int}`` become integer path parameters;
- wildcard (``*``) and websocket routes are skipped, since they cannot be
expressed in OpenAPI;
- method-agnostic routes (registered with ``app.route(path)``) are
documented under all standard HTTP methods;
- the mixin's own routes are excluded unless ``include_self`` is true.
Use :func:`~kaya.openapi.operation` to attach summaries, tags, request
bodies and response schemas to individual handlers; the handler docstring
is used as the operation description when no explicit one is given.
Example::
openapi = OpenAPIMixin(title='My API', version='1.0.0')
app = KayaApp(mixins=[openapi])
@app.GET('/users/${user_id:int}')
@operation(summary='Get a user', tags=['users'])
async def get_user(ctx: HttpContext, user_id: int) -> None:
...
"""
def __init__(self,
title: str,
version: str,
description: str = '',
spec_path: str = '/openapi.json',
docs_path: str = '/docs',
servers: Optional[Sequence[Mapping[str, Any]]] = None,
openapi_version: str = '3.1.0',
include_self: bool = False) -> None:
self._title = title
self._version = version
self._description = description
self._spec_path = spec_path
self._docs_path = docs_path
self._servers = servers
self._openapi_version = openapi_version
self._include_self = include_self
def apply(self, app: KayaApp) -> None:
@app.GET(self._spec_path)
async def openapi_spec(ctx: HttpContext) -> None:
spec = self.generate_spec(app)
await ctx.send_str(200,
json.dumps(spec, indent=2),
{'Content-Type': 'application/json'})
@app.GET(self._docs_path)
async def openapi_docs(ctx: HttpContext) -> None:
page = _DOCS_PAGE.format(title=escape(self._title), spec_path=self._spec_path)
await ctx.send_str(200, page, {'Content-Type': 'text/html; charset=utf-8'})
def generate_spec(self, app: KayaApp) -> Mapping[str, Any]:
"""Generate the OpenAPI document for ``app`` (also used by the
``spec_path`` endpoint on every request)."""
exclude_paths = frozenset() if self._include_self else frozenset((self._spec_path, self._docs_path))
return generate_spec(app,
title=self._title,
version=self._version,
description=self._description,
servers=self._servers,
exclude_paths=exclude_paths,
openapi_version=self._openapi_version)
@@ -0,0 +1,177 @@
from copy import deepcopy
from dataclasses import dataclass
from inspect import getdoc
from typing import AbstractSet, Any, Mapping, Optional, Sequence
from kaya.core import HttpMethod, KayaApp
from kaya.core._path_handler import PathHandler
from kaya.core._path_matcher import GlobMatcher, IntMatcher, Node, PathMatcher, StrMatcher
from ._metadata import get_metadata
#: HTTP methods documented for method-agnostic routes
#: (registered with ``app.route(path)`` without explicit methods).
STANDARD_METHODS: Sequence[HttpMethod] = (
HttpMethod.GET,
HttpMethod.PUT,
HttpMethod.POST,
HttpMethod.DELETE,
HttpMethod.OPTIONS,
HttpMethod.HEAD,
HttpMethod.PATCH,
)
@dataclass
class _Route:
raw_path: str
openapi_path: str
method: Optional[HttpMethod]
params: Sequence[Mapping[str, Any]]
handlers: Sequence[PathHandler]
def _walk(node: Node | PathMatcher,
raw_path: str,
openapi_path: str,
params: Sequence[Mapping[str, Any]],
routes: list[_Route]) -> None:
if node.handlers:
routes.append(_Route(raw_path or '/', openapi_path or '/', None, params, list(node.handlers)))
for key, child in node.children.items():
if isinstance(key, HttpMethod):
if key is not HttpMethod.WS and child.handlers:
routes.append(_Route(raw_path or '/', openapi_path or '/', key, params, list(child.handlers)))
else:
_walk(child, f'{raw_path}/{key}', f'{openapi_path}/{key}', params, routes)
for matcher in node.path_matchers:
if isinstance(matcher, GlobMatcher):
# Wildcard routes cannot be expressed in OpenAPI path syntax
continue
if isinstance(matcher, IntMatcher):
raw_segment = f'${{{matcher.name}:int}}'
schema: Mapping[str, Any] = {'type': 'integer'}
elif isinstance(matcher, StrMatcher):
raw_segment = f'${{{matcher.name}}}'
schema = {'type': 'string'}
else:
continue
param: dict[str, Any] = {
'name': matcher.name,
'in': 'path',
'required': True,
'schema': dict(schema),
}
_walk(matcher,
f'{raw_path}/{raw_segment}',
f'{openapi_path}/{{{matcher.name}}}',
(*params, param),
routes)
def _merge_parameters(params: Sequence[Mapping[str, Any]],
extra: Any) -> list[Mapping[str, Any]]:
merged: list[Mapping[str, Any]] = [dict(p) for p in params]
if not isinstance(extra, Sequence) or isinstance(extra, (str, bytes)):
return merged
for candidate in extra:
if not isinstance(candidate, Mapping):
continue
for i, existing in enumerate(merged):
if existing.get('name') == candidate.get('name') and existing.get('in') == candidate.get('in'):
merged[i] = candidate
break
else:
merged.append(candidate)
return merged
def _build_operation(params: Sequence[Mapping[str, Any]],
handler: PathHandler) -> Optional[dict[str, Any]]:
callback = getattr(handler, 'callback', None)
metadata = get_metadata(callback) if callback is not None else {}
if metadata.get('hidden'):
return None
operation: dict[str, Any] = {}
operation_id = metadata.get('operation_id')
if operation_id is not None:
operation['operationId'] = operation_id
summary = metadata.get('summary')
if summary is not None:
operation['summary'] = summary
description = metadata.get('description')
if description is None and callback is not None:
description = getdoc(callback)
if description is not None:
operation['description'] = description
tags = metadata.get('tags')
if tags is not None:
operation['tags'] = tags
if metadata.get('deprecated'):
operation['deprecated'] = True
parameters = _merge_parameters(params, metadata.get('parameters'))
if parameters:
operation['parameters'] = parameters
request_body = metadata.get('request_body')
if request_body is not None:
operation['requestBody'] = request_body
responses = metadata.get('responses')
operation['responses'] = responses if responses is not None else {
'default': {'description': 'Successful response'},
}
return operation
def generate_spec(app: KayaApp,
title: str,
version: str,
description: str = '',
servers: Optional[Sequence[Mapping[str, Any]]] = None,
exclude_paths: AbstractSet[str] = frozenset(),
openapi_version: str = '3.1.0') -> Mapping[str, Any]:
"""Generate an OpenAPI specification document from a :class:`KayaApp`.
The application's routing tree is walked and every route is converted to
an OpenAPI path item:
- static segments and ``${name}`` / ``${name:int}`` parameters are mapped
to OpenAPI path templating (``{name}``);
- wildcard (``*``) and websocket routes are skipped;
- method-agnostic routes are documented under all standard HTTP methods;
- routes whose Kaya path is in ``exclude_paths`` are skipped.
:param app: the application to inspect
:param title: value of ``info.title``
:param version: value of ``info.version``
:param description: value of ``info.description``
:param servers: list of OpenAPI server objects
:param exclude_paths: Kaya paths (e.g. ``/openapi.json``) to omit
:param openapi_version: OpenAPI version to declare
:return: the OpenAPI document as a JSON-serializable mapping
"""
routes: list[_Route] = []
_walk(app._tree.root, '', '', (), routes)
paths: dict[str, dict[str, Any]] = {}
for route in sorted(routes, key=lambda it: it.openapi_path):
if route.raw_path in exclude_paths or not route.handlers:
continue
operation = _build_operation(route.params, route.handlers[0])
if operation is None:
continue
path_item = paths.setdefault(route.openapi_path, {})
methods = STANDARD_METHODS if route.method is None else (route.method,)
for method in methods:
path_item[method.value.lower()] = deepcopy(operation)
info: dict[str, Any] = {'title': title, 'version': version}
if description:
info['description'] = description
spec: dict[str, Any] = {
'openapi': openapi_version,
'info': info,
'paths': paths,
}
if servers:
spec['servers'] = [dict(server) for server in servers]
return spec
+167
View File
@@ -0,0 +1,167 @@
import json
import unittest
from typing import Any, Mapping, Sequence
import httpx
from pwo import async_test
from kaya.core import HttpContext, HttpMethod, KayaApp, WebSocket
from kaya.openapi import OpenAPIMixin, operation
class OpenAPITest(unittest.TestCase):
app: KayaApp
def setUp(self) -> None:
self.app = KayaApp(mixins=[OpenAPIMixin(title='Test API', version='1.2.3')])
@self.app.GET('/hello')
async def hello(ctx: HttpContext) -> None:
"""Say hello."""
await ctx.send_str(200, 'Hello World!')
@self.app.GET('/users/${user_id:int}')
@operation(summary='Get a user',
tags=['users'],
responses={
200: {'description': 'The user'},
404: {'description': 'User not found'},
})
async def get_user(ctx: HttpContext, user_id: int) -> None:
await ctx.send_str(200, str(user_id))
@self.app.POST('/users/${name}')
async def create_user(ctx: HttpContext, name: str) -> None:
await ctx.send_str(201, name)
@self.app.route('/ping')
async def ping(ctx: HttpContext) -> None:
await ctx.send_str(200, 'pong')
@self.app.GET('/files/*', recursive=True)
async def serve_file(ctx: HttpContext, path: Sequence[str]) -> None:
await ctx.send_str(200, '/'.join(path))
@self.app.GET('/internal/health')
@operation(hidden=True)
async def health(ctx: HttpContext) -> None:
await ctx.send_str(200, 'ok')
@self.app.websocket('/echo')
async def echo(ws: WebSocket) -> None:
await ws.accept()
await ws.close()
async def _get_spec(self) -> Mapping[str, Any]:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
response = await client.get('/openapi.json')
self.assertEqual(200, response.status_code)
self.assertEqual('application/json', response.headers['Content-Type'])
return json.loads(response.text)
@async_test
async def test_spec_endpoint(self) -> None:
spec = await self._get_spec()
self.assertEqual('3.1.0', spec['openapi'])
self.assertEqual({'title': 'Test API', 'version': '1.2.3'}, spec['info'])
self.assertIn('paths', spec)
@async_test
async def test_static_route_with_docstring(self) -> None:
spec = await self._get_spec()
hello = spec['paths']['/hello']['get']
self.assertEqual('Say hello.', hello['description'])
self.assertIn('responses', hello)
@async_test
async def test_int_path_parameter(self) -> None:
spec = await self._get_spec()
operation = spec['paths']['/users/{user_id}']['get']
self.assertEqual('Get a user', operation['summary'])
self.assertEqual(['users'], operation['tags'])
self.assertEqual({
'200': {'description': 'The user'},
'404': {'description': 'User not found'},
}, operation['responses'])
self.assertEqual(
[{'name': 'user_id', 'in': 'path', 'required': True, 'schema': {'type': 'integer'}}],
operation['parameters'])
@async_test
async def test_str_path_parameter(self) -> None:
spec = await self._get_spec()
operation = spec['paths']['/users/{name}']['post']
self.assertEqual(
[{'name': 'name', 'in': 'path', 'required': True, 'schema': {'type': 'string'}}],
operation['parameters'])
@async_test
async def test_method_agnostic_route(self) -> None:
spec = await self._get_spec()
path_item = spec['paths']['/ping']
for method in ('get', 'put', 'post', 'delete', 'options', 'head', 'patch'):
self.assertIn(method, path_item)
@async_test
async def test_excluded_routes(self) -> None:
spec = await self._get_spec()
paths = spec['paths']
# wildcard routes cannot be expressed in OpenAPI
self.assertNotIn('/files/*', paths)
self.assertFalse(any('files' in path for path in paths))
# websocket routes are not part of OpenAPI
self.assertNotIn('/echo', paths)
# hidden operations are skipped
self.assertNotIn('/internal/health', paths)
# the mixin's own endpoints are excluded by default
self.assertNotIn('/openapi.json', paths)
self.assertNotIn('/docs', paths)
@async_test
async def test_docs_endpoint(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
response = await client.get('/docs')
self.assertEqual(200, response.status_code)
self.assertEqual('text/html; charset=utf-8', response.headers['Content-Type'])
self.assertIn('swagger-ui', response.text)
self.assertIn('/openapi.json', response.text)
@async_test
async def test_late_registered_routes_are_included(self) -> None:
@self.app.GET('/late')
async def late(ctx: HttpContext) -> None:
await ctx.send_str(200, 'late')
spec = await self._get_spec()
self.assertIn('/late', spec['paths'])
@async_test
async def test_custom_paths_and_self_inclusion(self) -> None:
app = KayaApp(mixins=[OpenAPIMixin(title='Custom',
version='0.1.0',
spec_path='/spec.json',
docs_path='/swagger',
include_self=True)])
@app.route('/items/${item_id:int}', HttpMethod.DELETE)
async def delete_item(ctx: HttpContext, item_id: int) -> None:
await ctx.send_empty(204)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
response = await client.get('/spec.json')
self.assertEqual(200, response.status_code)
spec = json.loads(response.text)
self.assertIn('delete', spec['paths']['/items/{item_id}'])
self.assertIn('/spec.json', spec['paths'])
self.assertIn('/swagger', spec['paths'])
response = await client.get('/swagger')
self.assertEqual(200, response.status_code)
self.assertIn('/spec.json', response.text)
if __name__ == '__main__':
unittest.main()
+3 -1
View File
@@ -141,7 +141,9 @@ class RsgiWebSocket(WebSocket):
.map(lambda it: (it[0], int(it[1])))
.or_else_throw(RuntimeError))
async def accept(self) -> None:
async def accept(self, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
# RSGI's websocket accept() takes no arguments (https://github.com/emmett-framework/granian/blob/master/docs/spec/RSGI.md),
# so handshake headers (e.g. Set-Cookie) cannot be sent on RSGI; they are silently ignored here.
self._transport = await self._protocol.accept()
async def receive(self) -> WebSocketMessage:
+52
View File
@@ -0,0 +1,52 @@
# kaya-session-memcache
Memcached-backed session storage for the Kaya web framework.
Provides `MemcacheSessionStore`, a `SessionStore` implementation (from
`kaya-session`) that persists session data in memcached via `aiomcache`, so
sessions are shared across processes and hosts.
## Usage
```python
import aiomcache
from kaya.core import KayaApp, HttpContext
from kaya.session import SessionMixin
from kaya.session.memcache import MemcacheSessionStore
client = aiomcache.Client('127.0.0.1', 11211)
session = SessionMixin(MemcacheSessionStore(client))
app = KayaApp(mixins=[session])
@app.GET('/')
async def home(ctx: HttpContext):
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
```
Sessions are stored under keys with the prefix `kaya:session:` (configurable
via the `prefix` argument). Server-side expiry uses memcached item expiration
and slides on each access when the session mixin passes a `max_age`. TTLs
larger than 30 days are automatically converted to absolute Unix timestamps,
as required by the memcached protocol.
## Serialization
Session data is serialized with `pickle` by default, so arbitrary Python
objects can be stored. A different serializer can be plugged in via the
`dumps`/`loads` arguments:
```python
import json
store = MemcacheSessionStore(
client,
dumps=lambda d: json.dumps(d).encode('utf-8'),
loads=lambda b: json.loads(b.decode('utf-8')),
)
```
**Warning:** pickle deserialization of untrusted data is unsafe. Only use the
default serializer with a trusted memcached server.
@@ -0,0 +1,57 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya-session-memcache"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "Memcached-backed session storage for the Kaya lightweight ASGI web framework"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'Programming Language :: Python :: 3',
]
dependencies = [
"kaya-session",
"aiomcache>=0.8",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
root = "../.."
version_file = "src/kaya/session/memcache/_version.py"
[tool.setuptools_scm.tag]
prefix = "release/"
@@ -0,0 +1,6 @@
from ._store import MemcacheSessionStore
__all__ = [
'MemcacheSessionStore',
]
@@ -0,0 +1,76 @@
import pickle
from time import time
from typing import Any, Callable, Optional
import aiomcache
from kaya.session import Session, SessionStore
_THIRTY_DAYS = 30 * 24 * 60 * 60
def _exptime(max_age: int, clock: Callable[[], float]) -> int:
"""Convert a relative TTL to a memcached ``exptime`` value.
Memcached interprets ``exptime`` values larger than 30 days as absolute
Unix timestamps, so large TTLs must be converted explicitly.
"""
if max_age > _THIRTY_DAYS:
return int(clock()) + max_age
return max_age
class MemcacheSessionStore(SessionStore):
"""Memcached-backed session store.
Suitable for multi-process and multi-host deployments: session data is
shared between all application instances connected to the same memcached
server.
Server-side expiry is delegated to memcached item expiration. When
``max_age`` is provided, active sessions slide the expiry window on each
access. TTLs larger than 30 days are converted to absolute Unix
timestamps, as required by the memcached protocol.
Session data is serialized with ``pickle`` by default, so arbitrary
Python objects can be stored. Custom serializers can be plugged in via
the ``dumps``/``loads`` arguments. Only connect this store to a trusted
memcached server, as pickle deserialization of untrusted data is unsafe.
"""
def __init__(
self,
client: aiomcache.Client,
prefix: str = "kaya:session:",
dumps: Callable[[dict[str, Any]], bytes] = pickle.dumps,
loads: Callable[[bytes], dict[str, Any]] = pickle.loads,
clock: Callable[[], float] = time,
) -> None:
self._client = client
self._prefix = prefix
self._dumps = dumps
self._loads = loads
self._clock = clock
def _key(self, session_id: str) -> bytes:
return f"{self._prefix}{session_id}".encode()
async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
key = self._key(session_id)
payload = await self._client.get(key)
if payload is None:
return None
if max_age is not None:
await self._client.touch(key, _exptime(max_age, self._clock))
data = self._loads(payload)
return Session(session_id, data)
async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
payload = self._dumps(dict(session))
exptime = _exptime(max_age, self._clock) if max_age is not None else 0
await self._client.set(self._key(session_id), payload, exptime=exptime)
async def delete(self, session_id: str) -> None:
await self._client.delete(self._key(session_id))
@@ -0,0 +1,203 @@
import unittest
from datetime import datetime, timezone
from typing import Optional
import httpx
from pwo import async_test
from kaya.core import KayaApp, HttpContext
from kaya.session import Session, SessionMixin
from kaya.session.memcache import MemcacheSessionStore
class FakeClock:
def __init__(self, start: float = 1_000_000.0) -> None:
self._now = start
def __call__(self) -> float:
return self._now
def advance(self, seconds: float) -> None:
self._now += seconds
class StubMemcacheClient:
"""In-memory stub implementing the aiomcache.Client subset used by the store."""
def __init__(self, clock: FakeClock) -> None:
self._clock = clock
self._data: dict[bytes, bytes] = {}
self._expires: dict[bytes, float] = {}
self.set_exptimes: list[int] = []
self.touch_exptimes: list[int] = []
def _expired(self, key: bytes) -> bool:
expires = self._expires.get(key)
return expires is not None and self._clock() > expires
async def get(self, key: bytes) -> Optional[bytes]:
if self._expired(key):
self._data.pop(key, None)
self._expires.pop(key, None)
return None
return self._data.get(key)
async def set(self, key: bytes, value: bytes, exptime: int = 0) -> bool:
self.set_exptimes.append(exptime)
self._data[key] = value
if exptime > 0:
self._expires[key] = self._clock() + exptime
else:
self._expires.pop(key, None)
return True
async def delete(self, key: bytes) -> bool:
existed = key in self._data
self._data.pop(key, None)
self._expires.pop(key, None)
return existed
async def touch(self, key: bytes, exptime: int) -> bool:
self.touch_exptimes.append(exptime)
if self._expired(key) or key not in self._data:
return False
self._expires[key] = self._clock() + exptime
return True
class MemcacheSessionStoreTest(unittest.TestCase):
clock: FakeClock
client: StubMemcacheClient
store: MemcacheSessionStore
def setUp(self) -> None:
self.clock = FakeClock()
self.client = StubMemcacheClient(self.clock)
self.store = MemcacheSessionStore(self.client, clock=self.clock) # type: ignore[arg-type]
@async_test
async def test_save_and_load_round_trip(self) -> None:
session = Session('abc', {'foo': 'bar', 'n': 42})
await self.store.save('abc', session)
loaded = await self.store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual('abc', loaded.id)
self.assertEqual({'foo': 'bar', 'n': 42}, dict(loaded))
@async_test
async def test_load_unknown_session_returns_none(self) -> None:
self.assertIsNone(await self.store.load('missing'))
@async_test
async def test_delete_removes_session(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}))
await self.store.delete('abc')
self.assertIsNone(await self.store.load('abc'))
@async_test
async def test_save_with_max_age_expires(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
self.assertIsNotNone(await self.store.load('abc'))
self.clock.advance(61)
self.assertIsNone(await self.store.load('abc'))
@async_test
async def test_save_without_max_age_never_expires(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertEqual([0], self.client.set_exptimes)
self.clock.advance(10_000_000)
self.assertIsNotNone(await self.store.load('abc'))
@async_test
async def test_load_slides_expiry_when_max_age_given(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
self.clock.advance(50)
loaded = await self.store.load('abc', max_age=60)
self.assertIsNotNone(loaded)
self.assertEqual([60], self.client.touch_exptimes)
self.clock.advance(50)
self.assertIsNotNone(await self.store.load('abc'))
@async_test
async def test_exptime_over_30_days_converted_to_absolute_timestamp(self) -> None:
max_age = 40 * 24 * 60 * 60
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=max_age)
self.assertEqual([int(self.clock()) + max_age], self.client.set_exptimes)
@async_test
async def test_exptime_exactly_30_days_stays_relative(self) -> None:
max_age = 30 * 24 * 60 * 60
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=max_age)
self.assertEqual([max_age], self.client.set_exptimes)
@async_test
async def test_touch_over_30_days_converted_to_absolute_timestamp(self) -> None:
max_age = 40 * 24 * 60 * 60
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=max_age)
self.clock.advance(100)
await self.store.load('abc', max_age=max_age)
self.assertEqual([int(self.clock()) + max_age], self.client.touch_exptimes)
@async_test
async def test_pickle_round_trip_of_non_json_values(self) -> None:
now = datetime(2026, 7, 20, 12, 0, 0, tzinfo=timezone.utc)
session = Session('abc', {'when': now, 'blob': b'\x00\x01', 'items': {1, 2, 3}})
await self.store.save('abc', session)
loaded = await self.store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(now, loaded['when'])
self.assertEqual(b'\x00\x01', loaded['blob'])
self.assertEqual({1, 2, 3}, loaded['items'])
@async_test
async def test_custom_prefix(self) -> None:
store = MemcacheSessionStore(self.client, prefix='myapp:sess:', clock=self.clock) # type: ignore[arg-type]
await store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertIn(b'myapp:sess:abc', self.client._data)
self.assertNotIn(b'kaya:session:abc', self.client._data)
@async_test
async def test_custom_serializer(self) -> None:
import json
store = MemcacheSessionStore(
self.client, # type: ignore[arg-type]
dumps=lambda d: json.dumps(d).encode('utf-8'),
loads=lambda b: json.loads(b.decode('utf-8')),
clock=self.clock,
)
await store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertEqual(b'{"foo": "bar"}', self.client._data[b'kaya:session:abc'])
loaded = await store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual({'foo': 'bar'}, dict(loaded))
class MemcacheSessionIntegrationTest(unittest.TestCase):
app: KayaApp
def setUp(self) -> None:
store = MemcacheSessionStore(StubMemcacheClient(FakeClock())) # type: ignore[arg-type]
self.app = KayaApp(mixins=[SessionMixin(store)])
@self.app.GET('/')
async def home(ctx: HttpContext) -> None:
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
@async_test
async def test_session_persists_across_requests(self) -> None:
transport = httpx.ASGITransport(app=self.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.assertEqual('visits: 1', r.text)
self.assertIn('Set-Cookie', r.headers)
r = await client.get('/')
self.assertEqual(200, r.status_code)
self.assertEqual('visits: 2', r.text)
+50
View File
@@ -0,0 +1,50 @@
# kaya-session-redis
Redis-backed session storage for the Kaya web framework.
Provides `RedisSessionStore`, a `SessionStore` implementation (from
`kaya-session`) that persists session data in Redis, so sessions are shared
across processes and hosts.
## Usage
```python
from redis.asyncio import Redis
from kaya.core import KayaApp, HttpContext
from kaya.session import SessionMixin
from kaya.session.redis import RedisSessionStore
client = Redis(host='localhost', port=6379)
session = SessionMixin(RedisSessionStore(client))
app = KayaApp(mixins=[session])
@app.GET('/')
async def home(ctx: HttpContext):
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
```
Sessions are stored under keys with the prefix `kaya:session:` (configurable
via the `prefix` argument). Server-side expiry uses Redis key TTLs and slides
on each access when the session mixin passes a `max_age`.
## Serialization
Session data is serialized with `pickle` by default, so arbitrary Python
objects can be stored. A different serializer can be plugged in via the
`dumps`/`loads` arguments:
```python
import json
store = RedisSessionStore(
client,
dumps=lambda d: json.dumps(d).encode('utf-8'),
loads=lambda b: json.loads(b.decode('utf-8')),
)
```
**Warning:** pickle deserialization of untrusted data is unsafe. Only use the
default serializer with a trusted Redis server.
@@ -0,0 +1,57 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya-session-redis"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "Redis-backed session storage for the Kaya lightweight ASGI web framework"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'Programming Language :: Python :: 3',
]
dependencies = [
"kaya-session",
"redis>=5.0",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine", "fakeredis"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
root = "../.."
version_file = "src/kaya/session/redis/_version.py"
[tool.setuptools_scm.tag]
prefix = "release/"
@@ -0,0 +1,6 @@
from ._store import RedisSessionStore
__all__ = [
'RedisSessionStore',
]
@@ -0,0 +1,57 @@
import pickle
from typing import Any, Callable, Optional, cast
from redis.asyncio import Redis
from kaya.session import Session, SessionStore
class RedisSessionStore(SessionStore):
"""Redis-backed session store.
Suitable for multi-process and multi-host deployments: session data is
shared between all application instances connected to the same Redis
server.
Server-side expiry is delegated to Redis key TTLs. When ``max_age`` is
provided, active sessions slide the expiry window on each access.
Session data is serialized with ``pickle`` by default, so arbitrary
Python objects can be stored. Custom serializers can be plugged in via
the ``dumps``/``loads`` arguments. Only connect this store to a trusted
Redis server, as pickle deserialization of untrusted data is unsafe.
"""
def __init__(
self,
client: Redis,
prefix: str = "kaya:session:",
dumps: Callable[[dict[str, Any]], bytes] = pickle.dumps,
loads: Callable[[bytes], dict[str, Any]] = pickle.loads,
) -> None:
self._client = client
self._prefix = prefix
self._dumps = dumps
self._loads = loads
def _key(self, session_id: str) -> str:
return f"{self._prefix}{session_id}"
async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
key = self._key(session_id)
payload = await self._client.get(key)
if payload is None:
return None
if max_age is not None:
await self._client.expire(key, max_age)
data = self._loads(cast(bytes, payload))
return Session(session_id, data)
async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
payload = self._dumps(dict(session))
await self._client.set(self._key(session_id), payload, ex=max_age)
async def delete(self, session_id: str) -> None:
await self._client.delete(self._key(session_id))
@@ -0,0 +1,123 @@
import unittest
from datetime import datetime, timezone
import fakeredis.aioredis
import httpx
from pwo import async_test
from kaya.core import KayaApp, HttpContext
from kaya.session import Session, SessionMixin
from kaya.session.redis import RedisSessionStore
class RedisSessionStoreTest(unittest.TestCase):
client: fakeredis.aioredis.FakeRedis
store: RedisSessionStore
def setUp(self) -> None:
self.client = fakeredis.aioredis.FakeRedis()
self.store = RedisSessionStore(self.client)
@async_test
async def test_save_and_load_round_trip(self) -> None:
session = Session('abc', {'foo': 'bar', 'n': 42})
await self.store.save('abc', session)
loaded = await self.store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual('abc', loaded.id)
self.assertEqual({'foo': 'bar', 'n': 42}, dict(loaded))
@async_test
async def test_load_unknown_session_returns_none(self) -> None:
self.assertIsNone(await self.store.load('missing'))
@async_test
async def test_delete_removes_session(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}))
await self.store.delete('abc')
self.assertIsNone(await self.store.load('abc'))
@async_test
async def test_save_sets_ttl_when_max_age_given(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
ttl = await self.client.ttl('kaya:session:abc')
self.assertGreater(ttl, 0)
self.assertLessEqual(ttl, 60)
@async_test
async def test_save_without_max_age_has_no_ttl(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertEqual(-1, await self.client.ttl('kaya:session:abc'))
@async_test
async def test_load_slides_expiry_when_max_age_given(self) -> None:
await self.store.save('abc', Session('abc', {'foo': 'bar'}), max_age=60)
await self.client.expire('kaya:session:abc', 10)
loaded = await self.store.load('abc', max_age=60)
self.assertIsNotNone(loaded)
ttl = await self.client.ttl('kaya:session:abc')
self.assertGreater(ttl, 10)
self.assertLessEqual(ttl, 60)
@async_test
async def test_pickle_round_trip_of_non_json_values(self) -> None:
now = datetime(2026, 7, 20, 12, 0, 0, tzinfo=timezone.utc)
session = Session('abc', {'when': now, 'blob': b'\x00\x01', 'items': {1, 2, 3}})
await self.store.save('abc', session)
loaded = await self.store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(now, loaded['when'])
self.assertEqual(b'\x00\x01', loaded['blob'])
self.assertEqual({1, 2, 3}, loaded['items'])
@async_test
async def test_custom_prefix(self) -> None:
store = RedisSessionStore(self.client, prefix='myapp:sess:')
await store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertIsNotNone(await self.client.get('myapp:sess:abc'))
self.assertIsNone(await self.client.get('kaya:session:abc'))
@async_test
async def test_custom_serializer(self) -> None:
import json
store = RedisSessionStore(
self.client,
dumps=lambda d: json.dumps(d).encode('utf-8'),
loads=lambda b: json.loads(b.decode('utf-8')),
)
await store.save('abc', Session('abc', {'foo': 'bar'}))
self.assertEqual(b'{"foo": "bar"}', await self.client.get('kaya:session:abc'))
loaded = await store.load('abc')
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual({'foo': 'bar'}, dict(loaded))
class RedisSessionIntegrationTest(unittest.TestCase):
app: KayaApp
def setUp(self) -> None:
store = RedisSessionStore(fakeredis.aioredis.FakeRedis())
self.app = KayaApp(mixins=[SessionMixin(store)])
@self.app.GET('/')
async def home(ctx: HttpContext) -> None:
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
@async_test
async def test_session_persists_across_requests(self) -> None:
transport = httpx.ASGITransport(app=self.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.assertEqual('visits: 1', r.text)
self.assertIn('Set-Cookie', r.headers)
r = await client.get('/')
self.assertEqual(200, r.status_code)
self.assertEqual('visits: 2', r.text)
+84
View File
@@ -0,0 +1,84 @@
# kaya-session
Session management for the Kaya web framework.
Provides server-side, identity-agnostic HTTP sessions via a session cookie. The
session data is accessible from request handlers as `ctx.session`.
## Usage
```python
from kaya.core import KayaApp, HttpContext
from kaya.session import SessionMixin, InMemorySessionStore
session = SessionMixin(InMemorySessionStore())
app = KayaApp(mixins=[session])
@app.GET('/')
async def home(ctx: HttpContext):
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
```
Sessions are created lazily: a cookie is only set when the handler modifies the
session.
`SessionMixin` is a `KayaMixin`, so the app stays a `KayaApp` and both ASGI and
RSGI keep working.
## WebSocket sessions
The same session is available in websocket handlers as `ws.session`:
```python
@app.websocket('/ws/visits')
async def ws_visits(ws: WebSocket):
await ws.accept()
visits = ws.session.get('visits', 0)
await ws.send_text(f'visits: {visits}')
```
The session is loaded from the cookie when the connection is opened and
persisted when the connection closes, if it was modified.
**Important:** although the ASGI spec allows custom headers on the WebSocket
handshake response (`websocket.accept` headers, spec 2.1+), most ASGI servers
in practice — including **Granian** and **Daphne** — do not forward them into
the HTTP `101` response. RSGI websocket handshakes cannot carry response headers
at all. Therefore, in real deployments a session cookie can only be set or
refreshed by an HTTP response. Use an HTTP endpoint to establish or update the
session before opening the WebSocket, and read the existing session in the
WebSocket handler.
## Session expiry
The cookie sent to the browser has a `Max-Age` (default 14 days), but that is
only a client-side hint. The real boundary is the store's server-side TTL,
which the mixin keeps in sync with the cookie `Max-Age`.
For `InMemorySessionStore`, a session expires if it is idle for longer than
`max_age`. Active sessions have their expiry slid forward on every access, so
a user that keeps visiting stays logged in. If the client ignores the cookie's
`Max-Age` and replays an old cookie value, the store rejects the expired
session and creates a fresh empty one.
Set `max_age=None` to disable server-side expiry (and the `Max-Age` cookie
attribute) entirely.
## Features
- `Session`: dict-like session object with modification tracking
- `SessionStore`: abstract store interface
- `InMemorySessionStore`: simple in-memory store for development/single-process
- `SessionMixin`: composable Kaya mixin managing session cookies and persistence
- Session ID regeneration (`session.regenerate_id()`) and invalidation
(`session.invalidate()`) for authentication layers
- WebSocket support: the session is exposed as `ws.session` in websocket
handlers, loaded at connect time and persisted on close
## Notes
- `InMemorySessionStore` does not survive process restarts and is not shared
across processes. Production deployments should use a store backed by a shared
storage system (planned).
+57
View File
@@ -0,0 +1,57 @@
[build-system]
requires = ["setuptools>=61.0", "setuptools-scm>=8"]
build-backend = "setuptools.build_meta"
[project]
name = "kaya-session"
dynamic = ["version"]
authors = [
{ name="Walter Oggioni", email="oggioni.walter@gmail.com" },
]
description = "Session management for the Kaya lightweight ASGI web framework"
readme = "README.md"
requires-python = ">=3.10"
license = "MIT"
classifiers = [
'Development Status :: 3 - Alpha',
'Topic :: Utilities',
'Intended Audience :: System Administrators',
'Intended Audience :: Developers',
'Environment :: Console',
'Programming Language :: Python :: 3',
]
dependencies = [
"kaya-core",
"pwo",
]
[project.optional-dependencies]
dev = [
"build", "mypy", "ipdb", "twine", "httpx", "httpx-ws"
]
[project.urls]
"Homepage" = "https://github.com/woggioni/kaya"
"Bug Tracker" = "https://github.com/woggioni/kaya/issues"
[tool.setuptools.packages.find]
where = ["src"]
namespaces = true
[tool.mypy]
python_version = "3.12"
disallow_untyped_defs = true
show_error_codes = true
no_implicit_optional = true
warn_return_any = true
warn_unused_ignores = true
exclude = ["scripts", "docs", "test"]
strict = true
[tool.setuptools_scm]
root = "../.."
version_file = "src/kaya/session/_version.py"
[tool.setuptools_scm.tag]
prefix = "release/"
@@ -0,0 +1,15 @@
from pkgutil import extend_path
__path__ = extend_path(__path__, __name__)
from ._mixin import SessionMixin
from ._session import Session
from ._store import InMemorySessionStore, SessionStore
__all__ = [
'InMemorySessionStore',
'Session',
'SessionMixin',
'SessionStore',
]
@@ -0,0 +1,37 @@
from http.cookies import SimpleCookie
from typing import Optional
def parse_cookie_value(header_value: str, cookie_name: str) -> Optional[str]:
"""Return the value of ``cookie_name`` from a ``Cookie`` header, if present."""
cookie = SimpleCookie()
cookie.load(header_value)
morsel = cookie.get(cookie_name)
if morsel is None:
return None
return morsel.value
def format_set_cookie(
name: str,
value: str,
path: str = '/',
max_age: Optional[int] = None,
httponly: bool = True,
secure: bool = False,
samesite: Optional[str] = 'Lax',
) -> str:
"""Return a ``Set-Cookie`` value string (without the header name)."""
cookie = SimpleCookie()
cookie[name] = value
morsel = cookie[name]
morsel['path'] = path
if max_age is not None:
morsel['max-age'] = max_age
if httponly:
morsel['httponly'] = True
if secure:
morsel['secure'] = True
if samesite is not None:
morsel['samesite'] = samesite
return morsel.OutputString()
@@ -0,0 +1,275 @@
from pathlib import Path
from typing import Any, AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Optional, Sequence
from kaya.core import HttpContext, KayaApp, KayaMixin, WebSocket
from kaya.core._types import StrOrStrings
from ._cookie import format_set_cookie, parse_cookie_value
from ._session import Session
from ._store import SessionStore
def _merge_cookie_header(
headers: Optional[Mapping[str, StrOrStrings]],
cookie_value: Optional[str],
) -> Optional[Mapping[str, StrOrStrings]]:
if cookie_value is None:
return headers
new_headers: dict[str, StrOrStrings] = dict(headers) if headers else {}
existing = new_headers.get('Set-Cookie')
if existing is None:
new_headers['Set-Cookie'] = cookie_value
elif isinstance(existing, str):
new_headers['Set-Cookie'] = (existing, cookie_value)
else:
new_headers['Set-Cookie'] = (*existing, cookie_value)
return new_headers
class SessionHttpContext(HttpContext):
"""HttpContext wrapper that exposes ``session`` and injects the session
cookie into response headers.
Works with any concrete ``HttpContext`` (ASGI or RSGI) because it only
relies on the abstract send methods, which all implementations share.
Attributes not explicitly overridden are delegated to the wrapped context
via ``__getattr__``, so protocol-specific fields (``pathsend``,
``receive``/``send`` for ASGI, ``protocol`` for RSGI, etc.) are passed
through transparently.
"""
def __init__(
self,
ctx: HttpContext,
session: Session,
cookie_injector: Callable[[], Optional[str]],
) -> None:
object.__setattr__(self, '_ctx', ctx)
object.__setattr__(self, 'session', session)
object.__setattr__(self, '_cookie_injector', cookie_injector)
def __getattr__(self, name: str) -> Any:
if name == '_ctx':
raise AttributeError(name)
return getattr(self._ctx, name)
def _inject_cookie(self, headers: Optional[Mapping[str, StrOrStrings]]) -> Optional[Mapping[str, StrOrStrings]]:
return _merge_cookie_header(headers, self._cookie_injector())
async def stream_body(self,
status: int,
body_generator: AsyncGenerator[bytes, None],
headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ctx.stream_body(status, body_generator, self._inject_cookie(headers))
async def send_bytes(self, status: int, body: bytes, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ctx.send_bytes(status, body, self._inject_cookie(headers))
async def send_str(self, status: int, body: str, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ctx.send_str(status, body, self._inject_cookie(headers))
async def send_file(self, status: int, path: Path, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ctx.send_file(status, path, self._inject_cookie(headers))
async def send_empty(self, status: int, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ctx.send_empty(status, self._inject_cookie(headers))
class SessionWebSocket(WebSocket):
"""WebSocket wrapper that exposes ``session`` and injects the session
cookie into the handshake response headers on ``accept()``.
Works with any concrete ``WebSocket`` (ASGI or RSGI) because it only
relies on the abstract methods, which all implementations share.
Attributes not explicitly overridden are delegated to the wrapped socket
via ``__getattr__``.
The cookie is only sent if the underlying transport supports handshake
response headers: ASGI does (spec version 2.1+), RSGI does not, so on
RSGI the session is still loaded and persisted but no cookie is set or
refreshed from a websocket connection.
"""
def __init__(
self,
ws: WebSocket,
session: Session,
cookie_injector: Callable[[], Optional[str]],
) -> None:
object.__setattr__(self, '_ws', ws)
object.__setattr__(self, 'session', session)
object.__setattr__(self, '_cookie_injector', cookie_injector)
def __getattr__(self, name: str) -> Any:
if name == '_ws':
raise AttributeError(name)
return getattr(self._ws, name)
async def accept(self, headers: Optional[Mapping[str, StrOrStrings]] = None) -> None:
await self._ws.accept(_merge_cookie_header(headers, self._cookie_injector()))
async def receive(self) -> Any:
return await self._ws.receive()
async def send_text(self, data: str) -> None:
await self._ws.send_text(data)
async def send_bytes(self, data: bytes) -> None:
await self._ws.send_bytes(data)
async def close(self, code: int = 1000) -> None:
await self._ws.close(code)
async def __anext__(self) -> Any:
return await self._ws.__anext__()
class _CookieInjector:
"""Computes the Set-Cookie value once (on first response) and caches it."""
def __init__(self, mixin: 'SessionMixin', session: Session) -> None:
self._mixin = mixin
self._session = session
self._value: Optional[str] = None
self._computed = False
def __call__(self) -> Optional[str]:
if not self._computed:
self._value = self._mixin._compute_cookie(self._session)
self._computed = True
return self._value
class SessionMixin(KayaMixin):
"""Kaya mixin providing server-side HTTP sessions.
Registers before/after request and websocket hooks that load and persist
the session, injecting the session cookie into HTTP responses via a
wrapped ``HttpContext`` and into websocket handshake responses via a
wrapped ``WebSocket``. Because the app stays a ``KayaApp``, both ASGI and
RSGI keep working.
For websockets the session is loaded when the connection is opened and
persisted when it closes if modified. The session cookie can only be set
or refreshed on the handshake response (ASGI only; RSGI websocket
handshakes cannot carry response headers).
Example::
session = SessionMixin(InMemorySessionStore())
app = KayaApp(mixins=[session])
@app.GET('/')
async def home(ctx: HttpContext):
ctx.session['visits'] = ctx.session.get('visits', 0) + 1
await ctx.send_str(200, f"visits: {ctx.session['visits']}")
"""
def __init__(
self,
store: SessionStore,
cookie_name: str = 'session_id',
path: str = '/',
max_age: Optional[int] = 14 * 24 * 60 * 60,
httponly: bool = True,
secure: bool = False,
samesite: Optional[str] = 'Lax',
) -> None:
self._store = store
self._cookie_name = cookie_name
self._path = path
self._max_age = max_age
self._httponly = httponly
self._secure = secure
self._samesite = samesite
def apply(self, app: KayaApp) -> None:
app.add_before_request_hook(self._before_request)
app.add_after_request_hook(self._after_request)
app.add_before_websocket_hook(self._before_websocket)
app.add_after_websocket_hook(self._after_websocket)
async def _before_request(self, ctx: HttpContext) -> Optional[HttpContext]:
session = await self._load_session(ctx.headers)
injector = _CookieInjector(self, session)
return SessionHttpContext(ctx, session, injector)
async def _after_request(self, ctx: HttpContext) -> None:
session = ctx.session
if not isinstance(session, Session):
return
await self._persist(session)
async def _before_websocket(self, ws: WebSocket) -> Optional[WebSocket]:
session = await self._load_session(ws.headers)
injector = _CookieInjector(self, session)
return SessionWebSocket(ws, session, injector)
async def _after_websocket(self, ws: WebSocket) -> None:
session = ws.session
if not isinstance(session, Session):
return
await self._persist(session)
async def _load_session(self, headers: Mapping[str, Sequence[str]]) -> Session:
session_id = self._extract_session_id(headers)
if session_id is not None:
loaded = await self._store.load(session_id, self._max_age)
if loaded is not None:
return loaded
return Session()
def _extract_session_id(self, headers: Mapping[str, Sequence[str]]) -> Optional[str]:
cookie_header_values = headers.get('cookie')
if cookie_header_values is None:
return None
if isinstance(cookie_header_values, str):
return parse_cookie_value(cookie_header_values, self._cookie_name)
for value in cookie_header_values:
found = parse_cookie_value(value, self._cookie_name)
if found is not None:
return found
return None
def _compute_cookie(self, session: Session) -> Optional[str]:
final_session_id = self._finalize_session_id(session)
if final_session_id is None:
return None
return format_set_cookie(
self._cookie_name,
final_session_id,
path=self._path,
max_age=0 if session.invalidated else self._max_age,
httponly=self._httponly,
secure=self._secure,
samesite=self._samesite,
)
def _finalize_session_id(self, session: Session) -> Optional[str]:
if session.invalidated:
return session.id
if session.id is None:
if session.modified or session.regenerate:
session.set_id(self._store.new_session_id())
elif session.regenerate:
session._old_id = session.id
session.set_id(self._store.new_session_id())
session._regenerate = False
return session.id
async def _persist(self, session: Session) -> None:
if session.invalidated:
old_id = session._old_id or session.id
if old_id is not None:
await self._store.delete(old_id)
return
if session._old_id is not None and session._old_id != session.id:
await self._store.delete(session._old_id)
session._old_id = None
if session.id is None and session.modified:
session.set_id(self._store.new_session_id())
if session.id is not None:
await self._store.save(session.id, session, self._max_age)
@@ -0,0 +1,82 @@
from typing import Any, Iterator, Mapping, MutableMapping, Optional
class Session(MutableMapping[str, Any]):
"""Dict-like session container with modification tracking.
The middleware uses the ``modified``, ``regenerate`` and ``invalidated``
flags to decide whether to persist the session, rotate its ID, or delete
it.
"""
def __init__(self, session_id: Optional[str] = None, data: Optional[Mapping[str, Any]] = None) -> None:
self._id: Optional[str] = session_id
self._old_id: Optional[str] = None
self._data: dict[str, Any] = dict(data) if data else {}
self._modified: bool = False
self._regenerate: bool = False
self._invalidated: bool = False
@property
def id(self) -> Optional[str]:
return self._id
def set_id(self, session_id: str) -> None:
self._id = session_id
@property
def modified(self) -> bool:
return self._modified
@property
def regenerate(self) -> bool:
return self._regenerate
@property
def invalidated(self) -> bool:
return self._invalidated
def regenerate_id(self) -> None:
"""Mark the session for ID rotation.
This is intended for authentication layers to defend against session
fixation: the middleware will create a new session ID, move the data to
it, and delete the old store entry.
"""
self._old_id = self._id
self._id = None
self._regenerate = True
self._modified = True
def invalidate(self) -> None:
"""Mark the session for deletion.
The middleware will clear the stored data and send an expired cookie.
"""
self._invalidated = True
self._modified = True
self._data.clear()
def mark_modified(self) -> None:
self._modified = True
def __getitem__(self, key: str) -> Any:
return self._data[key]
def __setitem__(self, key: str, value: Any) -> None:
self._data[key] = value
self._modified = True
def __delitem__(self, key: str) -> None:
del self._data[key]
self._modified = True
def __iter__(self) -> Iterator[str]:
return iter(self._data)
def __len__(self) -> int:
return len(self._data)
def clear(self) -> None:
self._data.clear()
self._modified = True
@@ -0,0 +1,85 @@
from abc import ABC, abstractmethod
from secrets import token_urlsafe
from time import monotonic
from typing import Any, Callable, Optional
from ._session import Session
class SessionStore(ABC):
"""Pluggable backend for session persistence."""
@abstractmethod
async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
"""Load an existing session, or return ``None`` if unknown or expired.
``max_age`` is the idle timeout in seconds. If provided, the store may
use it to enforce a server-side expiry and to slide the expiry window
on each access.
"""
pass
@abstractmethod
async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
"""Persist the session data under ``session_id``.
``max_age`` is the idle timeout in seconds. If provided, the store
should record the expiry time as ``now + max_age``.
"""
pass
@abstractmethod
async def delete(self, session_id: str) -> None:
"""Remove the session from the store."""
pass
def new_session_id(self) -> str:
"""Return a new opaque session identifier.
Subclasses may override this to use a backend-specific ID generator.
"""
return token_urlsafe(32)
class InMemorySessionStore(SessionStore):
"""Simple in-memory session store.
Suitable for development and single-process deployments. Session data is
lost when the process exits and is not shared between processes.
Sessions can optionally expire server-side after ``max_age`` seconds of
inactivity. Active sessions slide the expiry window on each access.
"""
def __init__(self, clock: Callable[[], float] = monotonic) -> None:
self._clock = clock
self._data: dict[str, dict[str, Any]] = {}
self._expires: dict[str, float] = {}
async def load(self, session_id: str, max_age: Optional[int] = None) -> Optional[Session]:
data = self._data.get(session_id)
if data is None:
return None
expires = self._expires.get(session_id)
now = self._clock()
if expires is not None and now > expires:
self._data.pop(session_id, None)
self._expires.pop(session_id, None)
return None
if max_age is not None and expires is not None:
self._expires[session_id] = now + max_age
return Session(session_id, data)
async def save(self, session_id: str, session: Session, max_age: Optional[int] = None) -> None:
self._data[session_id] = dict(session)
if max_age is not None:
self._expires[session_id] = self._clock() + max_age
else:
self._expires.pop(session_id, None)
async def delete(self, session_id: str) -> None:
self._data.pop(session_id, None)
self._expires.pop(session_id, None)
+331
View File
@@ -0,0 +1,331 @@
import asyncio
import unittest
from typing import Any
import httpx
from pwo import async_test
from kaya.core import KayaApp, HttpContext
from kaya.session import InMemorySessionStore, Session, SessionMixin, SessionStore
from kaya.session._cookie import format_set_cookie, parse_cookie_value
class FakeClock:
def __init__(self, start: float = 0.0) -> None:
self._now = start
def __call__(self) -> float:
return self._now
def advance(self, seconds: float) -> None:
self._now += seconds
class SessionTest(unittest.TestCase):
app: KayaApp
store: InMemorySessionStore
def setUp(self) -> None:
self.store = InMemorySessionStore()
self.app = KayaApp(mixins=[SessionMixin(self.store)])
@self.app.GET('/')
async def home(ctx: HttpContext) -> None:
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
@self.app.GET('/read')
async def read(ctx: HttpContext) -> None:
n = ctx.session.get('visits', 0)
await ctx.send_str(200, f'visits: {n}')
@self.app.GET('/write')
async def write(ctx: HttpContext) -> None:
ctx.session['foo'] = 'bar'
await ctx.send_str(200, 'ok')
@self.app.GET('/clear')
async def clear(ctx: HttpContext) -> None:
ctx.session.invalidate()
await ctx.send_str(200, 'cleared')
@self.app.GET('/rotate')
async def rotate(ctx: HttpContext) -> None:
ctx.session.regenerate_id()
await ctx.send_str(200, 'rotated')
@async_test
async def test_session_persists_across_requests(self) -> None:
transport = httpx.ASGITransport(app=self.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.assertEqual('visits: 1', r.text)
self.assertIn('Set-Cookie', r.headers)
r = await client.get('/')
self.assertEqual(200, r.status_code)
self.assertEqual('visits: 2', r.text)
@async_test
async def test_no_cookie_when_session_not_modified(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
r = await client.get('/read')
self.assertEqual(200, r.status_code)
self.assertEqual('visits: 0', r.text)
self.assertNotIn('Set-Cookie', r.headers)
@async_test
async def test_existing_session_refreshes_cookie(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
await client.get('/')
r = await client.get('/read')
self.assertEqual(200, r.status_code)
self.assertIn('Set-Cookie', r.headers)
self.assertEqual('visits: 1', r.text)
@async_test
async def test_cookie_attributes(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
r = await client.get('/')
set_cookie = r.headers['Set-Cookie']
self.assertIn('HttpOnly', set_cookie)
self.assertIn('SameSite=Lax', set_cookie)
self.assertIn('Path=/', set_cookie)
self.assertIn('Max-Age=', set_cookie)
@async_test
async def test_sessions_are_isolated(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
r1 = await client.get('/')
r2 = await client.get('/')
self.assertEqual('visits: 1', r1.text)
self.assertEqual('visits: 2', r2.text)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
r = await client.get('/')
self.assertEqual('visits: 1', r.text)
@async_test
async def test_invalidate(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
await client.get('/')
r = await client.get('/clear')
self.assertEqual(200, r.status_code)
self.assertEqual('cleared', r.text)
set_cookie = r.headers['Set-Cookie']
self.assertIn('Max-Age=0', set_cookie)
r = await client.get('/')
self.assertEqual('visits: 1', r.text)
@async_test
async def test_regenerate_id(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
await client.get('/')
cookies_before = {c.name: c.value for c in client.cookies.jar}
old_id = cookies_before.get('session_id')
self.assertIsNotNone(old_id)
self.assertIn(old_id, self.store._data)
r = await client.get('/rotate')
self.assertEqual(200, r.status_code)
self.assertEqual('rotated', r.text)
cookies_after = {c.name: c.value for c in client.cookies.jar}
new_id = cookies_after.get('session_id')
self.assertIsNotNone(new_id)
self.assertNotEqual(old_id, new_id)
self.assertNotIn(old_id, self.store._data)
self.assertIn(new_id, self.store._data)
r = await client.get('/read')
self.assertEqual('visits: 1', r.text)
@async_test
async def test_invalid_cookie_creates_fresh_session(self) -> None:
transport = httpx.ASGITransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://127.0.0.1:80') as client:
client.cookies.set('session_id', 'not-a-real-id')
r = await client.get('/')
self.assertEqual('visits: 1', r.text)
self.assertIn('Set-Cookie', r.headers)
@async_test
async def test_stale_cookie_cannot_access_old_data(self) -> None:
clock = FakeClock()
store = InMemorySessionStore(clock=clock)
app = KayaApp(mixins=[SessionMixin(store, max_age=60)])
@app.GET('/')
async def home(ctx: HttpContext) -> None:
ctx.session['secret'] = 'super-sensitive'
await ctx.send_str(200, 'ok')
@app.GET('/read')
async def read(ctx: HttpContext) -> None:
await ctx.send_str(200, ctx.session.get('secret', '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('/')
self.assertEqual('ok', r.text)
old_id = client.cookies['session_id']
self.assertIn(old_id, store._data)
self.assertIn(old_id, store._expires)
clock.advance(61)
r = await client.get('/read')
self.assertEqual('none', r.text)
self.assertNotIn(old_id, store._data)
self.assertNotIn(old_id, store._expires)
class SessionUnitTest(unittest.TestCase):
def test_session_is_dict_like(self) -> None:
session = Session()
session['a'] = 1
self.assertEqual(1, session['a'])
self.assertTrue('a' in session)
self.assertEqual({'a': 1}, dict(session))
self.assertTrue(session.modified)
def test_session_modification_tracking(self) -> None:
session = Session(data={'a': 1})
self.assertFalse(session.modified)
session['a'] = 2
self.assertTrue(session.modified)
def test_session_read_does_not_mark_modified(self) -> None:
session = Session(data={'a': 1})
self.assertFalse(session.modified)
_ = session['a']
self.assertFalse(session.modified)
def test_session_clear_marks_modified(self) -> None:
session = Session(data={'a': 1})
self.assertFalse(session.modified)
session.clear()
self.assertTrue(session.modified)
self.assertEqual(0, len(session))
def test_session_invalidate(self) -> None:
session = Session(session_id='abc', data={'a': 1})
self.assertFalse(session.modified)
session.invalidate()
self.assertTrue(session.invalidated)
self.assertTrue(session.modified)
self.assertEqual(0, len(session))
def test_session_regenerate_id(self) -> None:
session = Session(session_id='abc', data={'a': 1})
session.regenerate_id()
self.assertTrue(session.regenerate)
self.assertIsNone(session.id)
self.assertEqual('abc', session._old_id)
self.assertTrue(session.modified)
def test_session_store_is_abstract(self) -> None:
with self.assertRaises(TypeError):
SessionStore() # type: ignore[abstract]
def test_in_memory_store_round_trip(self) -> None:
store = InMemorySessionStore()
session_id = store.new_session_id()
session = Session(session_id, {'a': 1})
self.assertIsNone(store._data.get(session_id))
asyncio.run(store.save(session_id, session))
loaded = asyncio.run(store.load(session_id))
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(1, loaded['a'])
asyncio.run(store.delete(session_id))
self.assertIsNone(asyncio.run(store.load(session_id)))
def test_in_memory_store_expires_after_max_age(self) -> None:
clock = FakeClock()
store = InMemorySessionStore(clock=clock)
session_id = store.new_session_id()
asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
clock.advance(61)
self.assertIsNone(asyncio.run(store.load(session_id, max_age=60)))
self.assertIsNone(store._data.get(session_id))
self.assertIsNone(store._expires.get(session_id))
def test_in_memory_store_slides_expiry_on_load(self) -> None:
clock = FakeClock()
store = InMemorySessionStore(clock=clock)
session_id = store.new_session_id()
asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
clock.advance(30)
loaded = asyncio.run(store.load(session_id, max_age=60))
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(1, loaded['a'])
# Without sliding, the session would expire at t=60. With sliding it is now valid until t=90.
clock.advance(35)
loaded = asyncio.run(store.load(session_id, max_age=60))
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(1, loaded['a'])
def test_in_memory_store_no_expiry_without_max_age(self) -> None:
clock = FakeClock()
store = InMemorySessionStore(clock=clock)
session_id = store.new_session_id()
asyncio.run(store.save(session_id, Session(session_id, {'a': 1})))
clock.advance(1000000)
loaded = asyncio.run(store.load(session_id))
self.assertIsNotNone(loaded)
assert loaded is not None
self.assertEqual(1, loaded['a'])
def test_in_memory_store_invalidate_removes_expiry(self) -> None:
clock = FakeClock()
store = InMemorySessionStore(clock=clock)
session_id = store.new_session_id()
asyncio.run(store.save(session_id, Session(session_id, {'a': 1}), max_age=60))
asyncio.run(store.delete(session_id))
self.assertIsNone(store._data.get(session_id))
self.assertIsNone(store._expires.get(session_id))
class CookieUtilTest(unittest.TestCase):
def test_parse_cookie_value(self) -> None:
self.assertEqual('bar', parse_cookie_value('foo=bar; baz=qux', 'foo'))
self.assertEqual('qux', parse_cookie_value('foo=bar; baz=qux', 'baz'))
self.assertIsNone(parse_cookie_value('foo=bar', 'missing'))
self.assertIsNone(parse_cookie_value('', 'foo'))
def test_format_set_cookie(self) -> None:
value = format_set_cookie('sid', 'abc123', path='/', max_age=3600, httponly=True, secure=True, samesite='Strict')
self.assertIn('sid=abc123', value)
self.assertIn('Path=/', value)
self.assertIn('Max-Age=3600', value)
self.assertIn('HttpOnly', value)
self.assertIn('Secure', value)
self.assertIn('SameSite=Strict', value)
def test_format_set_cookie_without_secure(self) -> None:
value = format_set_cookie('sid', 'abc123', max_age=3600)
self.assertIn('sid=abc123', value)
self.assertIn('HttpOnly', value)
self.assertNotIn('Secure', value)
self.assertIn('SameSite=Lax', value)
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,194 @@
import unittest
import httpx
from pwo import async_test
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from kaya.core import KayaApp, HttpContext, WebSocket
from kaya.session import InMemorySessionStore, SessionMixin
class WebSocketSessionTest(unittest.TestCase):
app: KayaApp
store: InMemorySessionStore
def setUp(self) -> None:
self.store = InMemorySessionStore()
self.app = KayaApp(mixins=[SessionMixin(self.store)])
@self.app.GET('/')
async def home(ctx: HttpContext) -> None:
n = ctx.session.get('visits', 0) + 1
ctx.session['visits'] = n
await ctx.send_str(200, f'visits: {n}')
@self.app.GET('/read')
async def read(ctx: HttpContext) -> None:
await ctx.send_str(200, f"visits: {ctx.session.get('visits', 0)}"
f" ws_seen: {ctx.session.get('ws_seen', False)}")
@self.app.websocket('/visits')
async def ws_visits(ws: WebSocket) -> None:
await ws.accept()
await ws.send_text(f"visits: {ws.session.get('visits', 0)}")
@self.app.websocket('/mark')
async def ws_mark(ws: WebSocket) -> None:
await ws.accept()
ws.session['ws_seen'] = True
await ws.send_text('marked')
@self.app.websocket('/handshake-write')
async def ws_handshake_write(ws: WebSocket) -> None:
ws.session['ws_seen'] = True
await ws.accept()
await ws.send_text('marked')
@self.app.websocket('/peek')
async def ws_peek(ws: WebSocket) -> None:
await ws.accept()
await ws.send_text(f"visits: {ws.session.get('visits', 0)}")
@self.app.websocket('/rotate')
async def ws_rotate(ws: WebSocket) -> None:
old_id = ws.session.id
ws.session.regenerate_id()
await ws.accept()
await ws.send_text(f'old: {old_id} new: {ws.session.id}')
@self.app.websocket('/clear')
async def ws_clear(ws: WebSocket) -> None:
ws.session.invalidate()
await ws.accept()
await ws.send_text('cleared')
@async_test
async def test_ws_session_loaded_from_cookie(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
r = await client.get('/')
self.assertEqual('visits: 1', r.text)
async with aconnect_ws('/visits', client) as ws:
message = await ws.receive_text()
self.assertEqual('visits: 1', message)
@async_test
async def test_ws_session_persisted_on_close(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
await client.get('/')
async with aconnect_ws('/mark', client) as ws:
message = await ws.receive_text()
self.assertEqual('marked', message)
r = await client.get('/read')
self.assertEqual('visits: 1 ws_seen: True', r.text)
@async_test
async def test_ws_handshake_sets_cookie(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
self.assertNotIn('session_id', client.cookies)
async with aconnect_ws('/handshake-write', client) as ws:
message = await ws.receive_text()
self.assertEqual('marked', message)
self.assertIn('session_id', client.cookies)
self.assertIn(client.cookies['session_id'], self.store._data)
@async_test
async def test_ws_no_cookie_when_session_not_modified(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
async with aconnect_ws('/peek', client) as ws:
message = await ws.receive_text()
self.assertEqual('visits: 0', message)
self.assertNotIn('session_id', client.cookies)
@async_test
async def test_ws_session_unmodified_not_persisted(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
async with aconnect_ws('/peek', client) as ws:
message = await ws.receive_text()
self.assertEqual('visits: 0', message)
self.assertEqual(0, len(self.store._data))
@async_test
async def test_ws_session_new_session_saved_on_close(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
self.assertNotIn('session_id', client.cookies)
async with aconnect_ws('/mark', client) as ws:
message = await ws.receive_text()
self.assertEqual('marked', message)
self.assertEqual(1, len(self.store._data))
@async_test
async def test_ws_session_regenerate_id(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
await client.get('/')
old_id = client.cookies['session_id']
self.assertIn(old_id, self.store._data)
new_id = None
async with aconnect_ws('/rotate', client) as ws:
message = await ws.receive_text()
old_part, new_part = message.split(' new: ')
self.assertEqual(f'old: {old_id}', old_part)
new_id = new_part
self.assertNotEqual(old_id, new_id)
self.assertIsNotNone(new_id)
self.assertNotIn(old_id, self.store._data)
self.assertIn(new_id, self.store._data)
@async_test
async def test_ws_session_invalidate(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
await client.get('/')
old_id = client.cookies['session_id']
self.assertIn(old_id, self.store._data)
async with aconnect_ws('/clear', client) as ws:
message = await ws.receive_text()
self.assertEqual('cleared', message)
self.assertNotIn(old_id, self.store._data)
@async_test
async def test_ws_sessions_are_isolated(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
await client.get('/')
async with aconnect_ws('/visits', client) as ws:
message = await ws.receive_text()
self.assertEqual('visits: 1', message)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
async with aconnect_ws('/visits', client) as ws:
message = await ws.receive_text()
self.assertEqual('visits: 0', message)
@async_test
async def test_ws_invalid_cookie_creates_fresh_session(self) -> None:
transport = ASGIWebSocketTransport(app=self.app)
async with httpx.AsyncClient(transport=transport, base_url='http://testserver') as client:
client.cookies.set('session_id', 'not-a-real-id')
async with aconnect_ws('/mark', client) as ws:
message = await ws.receive_text()
self.assertEqual('marked', message)
self.assertNotIn('not-a-real-id', self.store._data)
self.assertEqual(1, len(self.store._data))
if __name__ == '__main__':
unittest.main()
+6
View File
@@ -1,6 +1,12 @@
kaya-core @ file:./packages/kaya-core
kaya-rsgi @ file:./packages/kaya-rsgi
kaya-session @ file:./packages/kaya-session
kaya-session-redis @ file:./packages/kaya-session-redis
kaya-session-memcache @ file:./packages/kaya-session-memcache
kaya-oidc @ file:./packages/kaya-oidc
kaya-openapi @ file:./packages/kaya-openapi
build
fakeredis
mypy
ipdb
twine
+34 -1
View File
@@ -7,6 +7,8 @@
--index-url https://gitea.woggioni.net/api/packages/woggioni/pypi/simple
--extra-index-url https://pypi.org/simple
aiomcache==0.8.2
# via kaya-session-memcache
anyio==4.14.2
# via
# httpx
@@ -29,7 +31,9 @@ charset-normalizer==3.4.9
click==8.4.2
# via granian
cryptography==49.0.0
# via secretstorage
# via
# pyjwt
# secretstorage
decorator==5.3.1
# via
# ipdb
@@ -38,6 +42,8 @@ docutils==0.23
# via readme-renderer
executing==2.2.1
# via stack-data
fakeredis==2.36.2
# via -r requirements-dev.in
granian==2.7.9
# via kaya-rsgi
h11==0.16.0
@@ -52,6 +58,7 @@ httpx==0.28.1
# via
# -r requirements-dev.in
# httpx-ws
# kaya-oidc
httpx-ws==0.9.0
# via -r requirements-dev.in
id==1.6.1
@@ -82,9 +89,26 @@ jeepney==0.9.0
file:./packages/kaya-core
# via
# -r requirements-dev.in
# kaya-oidc
# kaya-openapi
# kaya-rsgi
# kaya-session
file:./packages/kaya-oidc
# via -r requirements-dev.in
file:./packages/kaya-openapi
# via -r requirements-dev.in
file:./packages/kaya-rsgi
# via -r requirements-dev.in
file:./packages/kaya-session
# via
# -r requirements-dev.in
# kaya-oidc
# kaya-session-memcache
# kaya-session-redis
file:./packages/kaya-session-memcache
# via -r requirements-dev.in
file:./packages/kaya-session-redis
# via -r requirements-dev.in
keyring==25.7.0
# via twine
librt==0.13.0
@@ -127,6 +151,7 @@ pwo==0.1.2
# via
# kaya-core
# kaya-rsgi
# kaya-session
pycparser==3.0
# via cffi
pygments==2.20.0
@@ -135,10 +160,16 @@ pygments==2.20.0
# ipython-pygments-lexers
# readme-renderer
# rich
pyjwt[crypto]==2.13.0
# via kaya-oidc
pyproject-hooks==1.2.0
# via build
readme-renderer==45.0
# via twine
redis==8.0.1
# via
# fakeredis
# kaya-session-redis
requests==2.34.2
# via
# requests-toolbelt
@@ -151,6 +182,8 @@ rich==15.0.0
# via twine
secretstorage==3.5.0
# via keyring
sortedcontainers==2.4.0
# via fakeredis
stack-data==0.6.3
# via ipython
traitlets==5.15.1