From 28873f0607dd7cc7eb745e138deab70f465cf212 Mon Sep 17 00:00:00 2001 From: allen0099 Date: Tue, 29 Sep 2026 07:31:27 +0000 Subject: [PATCH] docs: include the examples/ files instead of copying code into the guides The guides now pull full-app code from examples/*.py with pymdownx.snippets (whole files or named sections), so the code shown is the code tests/test_examples.py runs. New examples cover the session basic and Redis apps and the JWT custom-claims guide. The docs workflow also runs on examples/ changes. --- .github/workflows/docs.yml | 2 + docs/BACKENDS.md | 23 +-- docs/DEVELOPMENT.md | 17 ++ docs/HTTP_CACHING.md | 39 ++--- docs/JWT_CLAIMS.md | 231 ++------------------------- docs/SESSION.md | 274 +++----------------------------- docs/STATE.md | 48 +----- examples/README.md | 13 +- examples/http_cache.py | 6 + examples/redis_backend.py | 2 + examples/session_api.py | 113 +++++++++++++ examples/session_jwt_claims.py | 196 +++++++++++++++++++++++ examples/session_redis.py | 171 ++++++++++++++++++++ i18n/zh-TW/GLOSSARY.md | 2 +- i18n/zh-TW/docs/BACKENDS.md | 23 +-- i18n/zh-TW/docs/HTTP_CACHING.md | 35 ++-- i18n/zh-TW/docs/JWT_CLAIMS.md | 231 ++------------------------- i18n/zh-TW/docs/SESSION.md | 267 ++----------------------------- i18n/zh-TW/docs/STATE.md | 48 +----- tests/test_examples.py | 137 +++++++++++++++- 20 files changed, 767 insertions(+), 1111 deletions(-) create mode 100644 examples/session_api.py create mode 100644 examples/session_jwt_claims.py create mode 100644 examples/session_redis.py diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml index 0744e10..0dc1a1c 100644 --- a/.github/workflows/docs.yml +++ b/.github/workflows/docs.yml @@ -11,6 +11,7 @@ on: - 'zensical.toml' - 'zensical.zh-TW.toml' - 'i18n/**' + - 'examples/**' - 'pyproject.toml' - 'uv.lock' - '.github/workflows/docs.yml' @@ -24,6 +25,7 @@ on: - 'zensical.toml' - 'zensical.zh-TW.toml' - 'i18n/**' + - 'examples/**' - 'pyproject.toml' - 'uv.lock' - '.github/workflows/docs.yml' diff --git a/docs/BACKENDS.md b/docs/BACKENDS.md index 9f877ef..6d7ed91 100644 --- a/docs/BACKENDS.md +++ b/docs/BACKENDS.md @@ -217,28 +217,11 @@ answers again. Every backend has `aclose()`, which releases what it holds open. Call it on shutdown, at the end of the FastAPI lifespan: + ```python -from contextlib import asynccontextmanager - -from fastapi import FastAPI - -from fastapi_cachex import BackendProxy -from fastapi_cachex.backends import AsyncRedisCacheBackend - - -@asynccontextmanager -async def lifespan(app: FastAPI): - backend = AsyncRedisCacheBackend(host="127.0.0.1", port=6379) - BackendProxy.set(backend) - try: - yield - finally: - BackendProxy.set(None) - await backend.aclose() - - -app = FastAPI(lifespan=lifespan) +--8<-- "examples/redis_backend.py:lifespan" ``` + The same lifespan works for every backend. A backend is also an async context manager, so `async with MemcachedBackend(servers=[...]) as backend:` closes it diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 2842442..d9f71c7 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -251,6 +251,23 @@ The **Docs** workflow (`.github/workflows/docs.yml`) and Read the Docs (`.readthedocs.yaml`) both run `zensical build --strict`, so a broken link, snippet path or docstring reference fails the PR. +A complete app shown in a guide is not copied into the page: it lives in +`examples/`, where `tests/test_examples.py` runs it, and the page includes it +inside a code fence, either the whole file or a named part: + +````markdown +```python +;--8<-- "examples/http_cache.py:routes" +``` +```` + +A part is marked in the example file with a start and an end comment (the +syntax is in [`examples/README.md`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/README.md)); +the site drops the marker lines. Short fragments that illustrate one call stay inline. The included code keeps +its English comments on the Traditional Chinese site too; explain it in the +text around the fence. `README.md` keeps its quick start inline, since PyPI and +GitHub render it without snippets. + ### Traditional Chinese translation A Traditional Chinese (`zh-TW`) translation is published at diff --git a/docs/HTTP_CACHING.md b/docs/HTTP_CACHING.md index 62c3f33..7399fac 100644 --- a/docs/HTTP_CACHING.md +++ b/docs/HTTP_CACHING.md @@ -8,30 +8,16 @@ Complete runnable example: [`examples/http_cache.py`](https://github.com/allen00 ## The `@cache` decorator + ```python -from fastapi import FastAPI -from fastapi_cachex import cache - -app = FastAPI() - - -@app.get("/") -@cache(ttl=60) # Cache for 60 seconds -async def read_root(): - return {"Hello": "World"} - - -@app.get("/no-cache") -@cache(no_cache=True) # Always revalidate: the handler runs on every request -async def non_cache_endpoint(): - return {"Hello": "World"} - - -@app.get("/no-store") -@cache(no_store=True) # Never store the response anywhere -async def non_store_endpoint(): - return {"Hello": "World"} +--8<-- "examples/http_cache.py:routes" ``` + + +`ttl=60` serves a stored response for 60 seconds, `no_cache=True` makes clients +revalidate every time, and `private=True` keeps a response out of the shared +backend. `no_store=True` keeps it out of every cache; all options are listed +under [Cache-Control directives](#cache-control-directives). Only GET requests are cached; other methods run the handler as usual. The handler does not need to declare a `Request` parameter — the decorator adds one @@ -738,5 +724,14 @@ add_routes( > a local or test app that should stay open, pass `dependencies=[]` to opt out > deliberately without the warning. +The runnable example guards them with a token from an environment variable, and +keeps them closed while the variable is unset: + + +```python +--8<-- "examples/http_cache.py:admin" +``` + + > [!NOTE] > On Memcached, which cannot enumerate keys, both routes return nothing. diff --git a/docs/JWT_CLAIMS.md b/docs/JWT_CLAIMS.md index 5f5b58e..716cb05 100644 --- a/docs/JWT_CLAIMS.md +++ b/docs/JWT_CLAIMS.md @@ -177,77 +177,11 @@ If your application needs additional JWT claims, write your own serializer and p The base class below does what the built-in `JWTTokenSerializer` does and leaves two hooks for the extra claims. It keeps its own copy of the settings, read from the public `SessionConfig` fields, instead of reaching into `JWTTokenSerializer`'s private attributes, which may change in any release. Like the built-in serializer, it follows `token.expires_at` in `to_string()`, so `exp` keeps up with sliding expiration. + ```python -from __future__ import annotations - -from datetime import datetime, timezone -from typing import Any - -import jwt - -from fastapi_cachex.session import SessionConfig -from fastapi_cachex.session.models import SessionToken - - -class CustomClaimsJWTSerializer: - """JWT serializer with the built-in claims plus extra ones from subclasses.""" - - # Claims that from_string() requires besides sid, iat and exp. - required_claims: tuple[str, ...] = () - - def __init__(self, config: SessionConfig) -> None: - self.secret = config.secret_key.get_secret_value() - self.algorithm = config.jwt_algorithm # must be HS256, HS384 or HS512 - self.issuer = config.jwt_issuer - self.audience = config.jwt_audience - self.leeway = config.jwt_leeway - self.session_ttl = config.session_ttl - - def extra_claims(self, token: SessionToken) -> dict[str, Any]: - """Return the claims to add to a new token.""" - return {} - - def check_claims(self, payload: dict[str, Any]) -> None: - """Raise ValueError if the extra claims of a verified token are wrong.""" - - def to_string(self, token: SessionToken) -> str: - """Encode a SessionToken as a signed JWT.""" - iat = int(token.issued_at.timestamp()) - if token.expires_at is not None: - exp = int(token.expires_at.timestamp()) - else: - exp = iat + self.session_ttl - - payload: dict[str, Any] = {"sid": token.session_id, "iat": iat, "exp": exp} - if self.issuer: - payload["iss"] = self.issuer - if self.audience: - payload["aud"] = self.audience - payload.update(self.extra_claims(token)) - return jwt.encode(payload, self.secret, algorithm=self.algorithm) - - def from_string(self, token_str: str) -> SessionToken: - """Verify a JWT and turn it back into a SessionToken.""" - try: - payload = jwt.decode( - token_str, - self.secret, - algorithms=[self.algorithm], - issuer=self.issuer, - audience=self.audience, - leeway=self.leeway, - options={"require": ["sid", "iat", "exp", *self.required_claims]}, - ) - except jwt.InvalidTokenError as e: - msg = "Invalid JWT token" - raise ValueError(msg) from e - - self.check_claims(payload) - issued_at = datetime.fromtimestamp(int(payload["iat"]), tz=timezone.utc) - return SessionToken( - session_id=str(payload["sid"]), signature="", issued_at=issued_at - ) +--8<-- "examples/session_jwt_claims.py:serializer" ``` + PyJWT verifies the signature, `exp`, `iat` and (when present) `nbf` by default, and `iss`/`aud` when `issuer`/`audience` are given. Two checks of the built-in serializer are not repeated here: it rejects an asymmetric `jwt_algorithm` and warns about a `secret_key` shorter than the HMAC output. This class signs with `secret_key` too, so keep an `HS*` algorithm; for an asymmetric one, hold the private and public keys in the class and use them in `jwt.encode()` and `jwt.decode()`. @@ -274,83 +208,26 @@ class ExtendedJWTSerializer(CustomClaimsJWTSerializer): ### Example 2: Adding Multi-Tenant Custom Claims + ```python -from typing import Any - -from fastapi_cachex.session import SessionConfig -from fastapi_cachex.session.models import SessionToken - - -class MultiTenantJWTSerializer(CustomClaimsJWTSerializer): - """Adds tenant_id and api_version, and rejects tokens for other tenants.""" - - required_claims = ("tenant_id", "api_version") - - def __init__( - self, config: SessionConfig, tenant_id: str, api_version: str = "v1" - ) -> None: - super().__init__(config) - self.tenant_id = tenant_id - self.api_version = api_version - - def extra_claims(self, token: SessionToken) -> dict[str, Any]: - return {"tenant_id": self.tenant_id, "api_version": self.api_version} - - def check_claims(self, payload: dict[str, Any]) -> None: - if payload["tenant_id"] != self.tenant_id: - msg = f"Invalid tenant_id: expected {self.tenant_id}, got {payload['tenant_id']}" - raise ValueError(msg) - if payload["api_version"] != self.api_version: - msg = f"Unsupported API version: {payload['api_version']}" - raise ValueError(msg) +--8<-- "examples/session_jwt_claims.py:multi-tenant" ``` + ### Using a Custom Serializer #### Option 1: Pass it to `SessionManager` (recommended) + ```python -from fastapi import FastAPI - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, -) - -app = FastAPI() - -# Set up the backend and config -backend = AsyncRedisCacheBackend(host="localhost", port=6379) -config = SessionConfig( - secret_key="your-secret-key-at-least-32-characters", - token_format="jwt", - jwt_algorithm="HS256", - jwt_issuer="your-company", - jwt_audience="your-api", -) - -# Create the custom serializer -custom_serializer = MultiTenantJWTSerializer( - config=config, - tenant_id="acme-corp", - api_version="v2", -) - -# Initialize the SessionManager -manager = SessionManager(backend, config, token_serializer=custom_serializer) - -# Add the middleware -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=manager, - config=config, -) +--8<-- "examples/session_jwt_claims.py:setup" ``` + When `token_serializer` is given it overrides the built-in choice made from `token_format`. Keep `token_format="jwt"` anyway: with `"simple"`, `SessionManager` additionally performs its own HMAC signature check on the parsed token, which a JWT-based serializer does not provide. +The example uses `MemoryBackend` so it runs without a server; any backend works, for example the Redis setup of [Backends](BACKENDS.md#closing-a-backend). + #### Option 2: Subclass `SessionManager` (advanced) ```python @@ -381,89 +258,13 @@ Do not replace the serializer by assigning a private attribute after constructio ## Complete Application Example -```python -from __future__ import annotations - -from fastapi import Depends, FastAPI, HTTPException -from pydantic import BaseModel - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - Session, - SessionConfig, - SessionManager, - SessionUser, - require_user_session, -) - -# Uses the MultiTenantJWTSerializer defined above +[`examples/session_jwt_claims.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_jwt_claims.py) puts the pieces above together into a runnable app: -app = FastAPI() - -# Initialization -backend = AsyncRedisCacheBackend(host="localhost", port=6379) -config = SessionConfig( - secret_key="your-secret-key-min-32-chars-long!!", - token_format="jwt", - jwt_algorithm="HS256", - jwt_issuer="acme-corp", - jwt_audience="acme-api", -) - -# Create the custom serializer -serializer = MultiTenantJWTSerializer( - config=config, - tenant_id="acme-corp", - api_version="v2", -) - -manager = SessionManager(backend, config, token_serializer=serializer) - -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=manager, - config=config, -) - - -class LoginRequest(BaseModel): - username: str - password: str - - -@app.post("/auth/login") -async def login(credentials: LoginRequest) -> dict[str, str]: - """Login endpoint that returns a JWT containing tenant_id.""" - # Authenticate the user (omitted) - if credentials.username != "admin": - raise HTTPException(status_code=401, detail="Invalid credentials") - - user = SessionUser(user_id="123", username=credentials.username) - session, token = await manager.create_session(user=user) - - # The token now contains the tenant_id and api_version claims - return { - "token": token, - "token_type": "bearer", - "tenant_id": "acme-corp", # Could also be read from configuration - } - - -@app.get("/api/profile") -async def get_profile( - session: Session = Depends(require_user_session), -) -> dict[str, str | None]: - """Protected endpoint; tenant_id is validated automatically.""" - # tenant_id and api_version were already validated while decoding the JWT. - # A token for another tenant fails to decode, so the middleware loads no - # session and require_user_session responds with 401. - assert session.user is not None - return { - "user_id": session.user.user_id, - "username": session.user.username, - } + +```python +--8<-- "examples/session_jwt_claims.py" ``` + ## Security Considerations diff --git a/docs/SESSION.md b/docs/SESSION.md index dcfc9bb..050f1ac 100644 --- a/docs/SESSION.md +++ b/docs/SESSION.md @@ -60,103 +60,23 @@ uv add "fastapi-cachex[jwt]" ### 2. Basic Usage -```python -from fastapi import Depends, FastAPI, HTTPException -from pydantic import BaseModel - -from fastapi_cachex.backends import MemoryBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, - SessionUser, - get_optional_session, - get_session, -) -from fastapi_cachex.session.dependencies import AuthenticatedSession - -# Create the FastAPI application -app = FastAPI() - -# Session configuration (API-first architecture: the client manages the token) -config = SessionConfig( - secret_key="your-secret-key-min-32-chars-long!!!", # at least 32 characters - session_ttl=3600, # 1 hour -) - -# Set up the backend and the session manager -backend = MemoryBackend() -session_manager = SessionManager(backend, config) - -# Add the session middleware (SessionMiddleware is deprecated and removed in 0.4.0) -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=session_manager, - config=config, -) - -# Alternatively, register the manager on the proxy instead of passing it in: -# -# from fastapi_cachex.session import SessionManagerProxy -# -# SessionManagerProxy.set(session_manager) -# app.add_middleware(FastAPICacheXSessionMiddleware) # picked up from the proxy -# -# When `config` is omitted, the middleware uses `session_manager.config`. +This is [`examples/session_api.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_api.py): an API client logs in, +keeps the token it gets back and sends it on later requests. + +```python +--8<-- "examples/session_api.py" +``` + -# Credentials arrive in the JSON request body, never in the query string, -# which ends up in browser history and access logs. -class LoginRequest(BaseModel): - username: str - password: str +Instead of passing the manager to the middleware, you can register it on the +proxy. When `config` is omitted, the middleware uses `session_manager.config`: +```python +from fastapi_cachex.session import SessionManagerProxy -# Login endpoint -@app.post("/login") -async def login(credentials: LoginRequest): - # Authenticate the user (simplified here) - if credentials.username != "admin" or credentials.password != "secret": - raise HTTPException(status_code=401, detail="Invalid credentials") - - # Create the session - user = SessionUser( - user_id="123", - username=credentials.username, - roles=["admin"], - ) - session, token = await session_manager.create_session(user=user) - - # Return the token for the client to store (localStorage/sessionStorage). - # The client sends it on later requests in the Authorization or X-Session-Token header. - return {"message": "Login successful", "token": token} - - -# Endpoint that requires a logged-in user -@app.get("/profile") -async def get_profile(session: AuthenticatedSession): - """Requires a session with a user; 401 otherwise, anonymous sessions included.""" - return { - "user_id": session.user.user_id, - "username": session.user.username, - "roles": session.user.roles, - } - - -# Endpoint with optional authentication -@app.get("/public") -async def public_endpoint(session=Depends(get_optional_session)): - """Accessible with or without a session.""" - if session and session.user: - return {"message": f"Hello, {session.user.username}!"} - return {"message": "Hello, guest!"} - - -# Logout endpoint -@app.post("/logout") -async def logout(session=Depends(get_session)): - await session_manager.delete_session(session.session_id) - return {"message": "Logged out"} +SessionManagerProxy.set(session_manager) +app.add_middleware(FastAPICacheXSessionMiddleware) # picked up from the proxy ``` `get_session` (and its alias `require_session`) raises `401 Authentication required` with a @@ -172,8 +92,9 @@ anonymous, so `session.user` is `None`. that anyone logged in. Any visitor who reaches a route that writes to `request.session` (a cart, a CSRF value) gets one. Guard routes that need a logged-in user with `require_user_session` (or its annotated form `AuthenticatedSession`), which also answers `401` when `session.user` is -`None`, as `/profile` above does. `/logout` only deletes the session, so `get_session` is enough -there. +`None`, as `/profile` above does. `/logout` only deletes the session, so `SessionDep` (the +annotated form of `get_session`) is enough there, and `/public` uses `OptionalSession` +(`get_optional_session`), which gives `None` instead of answering `401`. `UserSessionDep` does not check for a user despite its name; it is an alias of `SessionDep` until 0.4.0, which is planned to make it require one. @@ -190,164 +111,15 @@ such a session. ### 3. Full Example (Redis Backend) -```python -from datetime import datetime, timezone - -from fastapi import Depends, FastAPI, HTTPException, Request -from pydantic import BaseModel - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, - SessionUser, - get_session, -) -from fastapi_cachex.session.dependencies import AuthenticatedSession, ClientIPDep - -app = FastAPI() - -# Redis backend -backend = AsyncRedisCacheBackend( - host="localhost", - port=6379, - db=0, -) - -# Session configuration with security options -config = SessionConfig( - secret_key="your-very-secret-key-at-least-32-characters-long!!", - session_ttl=3600, - sliding_expiration=True, - sliding_threshold=0.5, - ip_binding=True, # enable IP binding - user_agent_binding=False, # UA binding (optional) -) - -session_manager = SessionManager(backend, config) - -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=session_manager, - config=config, -) - - -class LoginRequest(BaseModel): - username: str - password: str - - -@app.post("/api/auth/login") -async def login(credentials: LoginRequest, request: Request, client_ip: ClientIPDep): - # Authenticate the user (should query a database) - username = credentials.username - if not authenticate_user(username, credentials.password): - raise HTTPException(status_code=401, detail="Invalid credentials") - - # Create the session - user = SessionUser( - user_id=get_user_id(username), - username=username, - email=f"{username}@example.com", - roles=get_user_roles(username), - ) - - # Collect client information for the bindings. `client_ip` is the address - # the middleware checks later, including behind trusted proxies. - user_agent = request.headers.get("user-agent") - - session, token = await session_manager.create_session( - user=user, - ip_address=client_ip, - user_agent=user_agent, - ) - - # Add a flash message - session.add_flash_message("Login successful!", "success") - await session_manager.update_session(session) - - return { - "message": "Login successful", - "token": token, # the client stores this token and sends it on later requests - "user": { - "username": user.username, - "roles": user.roles, - }, - } - +This is [`examples/session_redis.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_redis.py). Like +[`examples/redis_backend.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/redis_backend.py), it reads the Redis +settings from `REDIS_HOST`, `REDIS_PORT`, `REDIS_DB` and `REDIS_PASSWORD`. -@app.get("/api/user/profile") -async def get_user_profile(session: AuthenticatedSession): - """Return the user's profile (requires a logged-in user).""" - return { - "user_id": session.user.user_id, - "username": session.user.username, - "email": session.user.email, - "roles": session.user.roles, - "session_created": session.created_at.isoformat(), - "last_accessed": session.last_accessed.isoformat(), - } - - -@app.post("/api/user/update") -async def update_user_profile( - email: str, - session: AuthenticatedSession, -): - """Update the user's profile.""" - session.user.email = email - session.data["last_updated"] = datetime.now(timezone.utc).isoformat() - - # Persist the updated session - await session_manager.update_session(session) - - return {"message": "Profile updated"} - - -@app.get("/api/messages") -async def get_flash_messages(session=Depends(get_session)): - """Return and clear the flash messages.""" - messages = session.get_flash_messages(clear=True) - # Clearing only changes the in-memory object; save it so the messages - # are not shown again on the next request. - await session_manager.update_session(session) - return {"messages": messages} - - -@app.post("/api/auth/logout") -async def logout(session=Depends(get_session)): - """Log out.""" - await session_manager.delete_session(session.session_id) - - # The client should discard its stored token - return {"message": "Logged out successfully"} - - -@app.post("/api/auth/logout-all") -async def logout_all_devices(session: AuthenticatedSession): - """Log out from all devices.""" - user_id = session.user.user_id - count = await session_manager.delete_user_sessions(user_id) - return {"message": f"Logged out from {count} devices"} - - -# Helper functions (illustrative only) -def authenticate_user(username: str, password: str) -> bool: - # A real implementation queries the database and verifies the password hash - return True - - -def get_user_id(username: str) -> str: - # A real implementation reads this from the database - return f"user_{username}" - - -def get_user_roles(username: str) -> list[str]: - # A real implementation reads this from the database - return ["user"] if username != "admin" else ["admin", "user"] + +```python +--8<-- "examples/session_redis.py" ``` + Changes made to a `Session` object inside a handler (flash messages, `session.data`, `session.user`) are only persisted when you call `session_manager.update_session(session)`. diff --git a/docs/STATE.md b/docs/STATE.md index cbc7902..e2bab91 100644 --- a/docs/STATE.md +++ b/docs/STATE.md @@ -19,55 +19,15 @@ does remove them, because it clears everything under the backend's namespace. Everything in this guide can also be imported from the top-level `fastapi_cachex` package. -Complete runnable example: [`examples/oauth_state.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/oauth_state.py). +The quick start below is the complete, runnable [`examples/oauth_state.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/oauth_state.py). ## Quick start + ```python -import secrets - -from fastapi import FastAPI, HTTPException, Request -from fastapi.responses import RedirectResponse - -from fastapi_cachex import BackendProxy -from fastapi_cachex.backends import MemoryBackend -from fastapi_cachex.state import StateError, StateManagerDep - -app = FastAPI() -BackendProxy.set(MemoryBackend()) - -BINDING_COOKIE = "oauth_binding" - - -@app.get("/login") -async def login(states: StateManagerDep): - nonce = secrets.token_urlsafe(32) - state = await states.create_state(binding=nonce, metadata={"next": "/dashboard"}) - response = RedirectResponse( - f"https://provider.example.com/authorize?state={state}&client_id=..." - ) - # Lax, not Strict: the callback is a cross-site navigation from the provider. - response.set_cookie( - BINDING_COOKIE, nonce, max_age=600, httponly=True, secure=True, samesite="lax" - ) - return response - - -@app.get("/callback") -async def callback(request: Request, state: str, code: str, states: StateManagerDep): - try: - # One-time: deleted on retrieval. Rejected unless this browser started the flow. - data = await states.consume_state( - state, binding=request.cookies.get(BINDING_COOKIE) - ) - except StateError as e: # unknown, expired, malformed or issued to another browser - raise HTTPException(status_code=400, detail="Invalid state") from e - - # Exchange the code for tokens, create a session ... - response = RedirectResponse(data.metadata.get("next", "/")) - response.delete_cookie(BINDING_COOKIE) - return response +--8<-- "examples/oauth_state.py" ``` + If the provider posts the callback (`response_mode=form_post`), a `SameSite=Lax` cookie is not sent with that cross-site POST; use `samesite="none"` (which requires `secure=True`) diff --git a/examples/README.md b/examples/README.md index ea5ffec..02df51e 100644 --- a/examples/README.md +++ b/examples/README.md @@ -2,14 +2,17 @@ Each file here is a complete FastAPI app that shows one feature of FastAPI-CacheX. They use only the public API and the in-memory backend (except -`redis_backend.py`), so they run without any server. +`redis_backend.py` and `session_redis.py`), so they run without any server. | Example | What it shows | Needs | |---------|---------------|-------| | [`http_cache.py`](http_cache.py) | `@cache` with a TTL, ETag and `304 Not Modified`, `no_cache` and `private` routes, `clear_path()` after an update, the monitoring routes behind an admin check | — | | [`app_cache.py`](app_cache.py) | `CacheManager.get_or_set()` for an expensive call, `add()` as an idempotency check, the `AppCache` dependency | — | | [`session_login.py`](session_login.py) | `FastAPICacheXSessionMiddleware` with cookies: an anonymous session, login with `login()`, `AuthenticatedSession`, logout with `request.session.clear()` | — | +| [`session_api.py`](session_api.py) | Sessions for an API client that keeps its own token: returned by `/login`, sent back as `Authorization: Bearer` or `X-Session-Token`; `AuthenticatedSession`, `OptionalSession`, logout with `delete_session()` | — | +| [`session_redis.py`](session_redis.py) | Sessions on Redis with IP binding and sliding expiration, flash messages, `update_session()` and logout on every device with `delete_user_sessions()` | `redis` extra, a Redis server | | [`session_jwt.py`](session_jwt.py) | Sessions with `token_format="jwt"` for API clients (`Authorization: Bearer`), revoked on logout | `jwt` extra | +| [`session_jwt_claims.py`](session_jwt_claims.py) | A custom JWT serializer passed as `token_serializer`, adding `tenant_id` and `api_version` claims and rejecting other tenants' tokens | `jwt` extra | | [`oauth_state.py`](oauth_state.py) | One-time OAuth `state` values with `StateManager`, bound to the browser that started the flow | — | | [`cache_lock.py`](cache_lock.py) | `CacheLock`, waiting (`async with`) and non-blocking (`acquire(blocking=False)`) | — | | [`rate_limit.py`](rate_limit.py) | A fixed-window rate limiter on `backend.increment()`, answering `429` with `Retry-After` | — | @@ -32,8 +35,12 @@ a dependency of this project, hence `--with`. Uvicorn works too: uv run --with uvicorn uvicorn examples.http_cache:app --reload ``` -`redis_backend.py` reads `REDIS_HOST`, `REDIS_PORT`, `REDIS_DB` and -`REDIS_PASSWORD`; point it at a server you can write to. +`redis_backend.py` and `session_redis.py` read `REDIS_HOST`, `REDIS_PORT`, +`REDIS_DB` and `REDIS_PASSWORD`; point them at a server you can write to. + +Some files mark parts with `# --8<-- [start:name]` / `# --8<-- [end:name]` +comments. The documentation site includes those parts (or the whole file) in +its guides, so the code shown there is the code tested here. In your own project, install the extras an example needs, for example `uv add "fastapi-cachex[jwt]"`, and copy the file. diff --git a/examples/http_cache.py b/examples/http_cache.py index 73c9954..4fdbf54 100644 --- a/examples/http_cache.py +++ b/examples/http_cache.py @@ -9,6 +9,7 @@ uv run --with "fastapi-cli[standard]" fastapi dev examples/http_cache.py """ +# --8<-- [start:routes] import os import secrets from collections.abc import AsyncIterator @@ -87,6 +88,10 @@ async def my_preferences() -> dict[str, str]: return {"theme": "dark"} +# --8<-- [end:routes] + + +# --8<-- [start:admin] def require_admin(x_admin_token: str | None = Header(default=None)) -> None: """Allow the monitoring routes only with the token from ``CACHE_ADMIN_TOKEN``.""" expected = os.environ.get("CACHE_ADMIN_TOKEN") @@ -100,3 +105,4 @@ def require_admin(x_admin_token: str | None = Header(default=None)) -> None: # GET /_cache/cached-hits and /_cache/cached-records. They have no auth of # their own, so always pass a dependency that has. add_routes(app, prefix="/_cache", dependencies=[Depends(require_admin)]) +# --8<-- [end:admin] diff --git a/examples/redis_backend.py b/examples/redis_backend.py index cbf0c9b..ce9b7d8 100644 --- a/examples/redis_backend.py +++ b/examples/redis_backend.py @@ -12,6 +12,7 @@ REDIS_PORT=6379 uv run --with "fastapi-cli[standard]" fastapi dev examples/redis_backend.py """ +# --8<-- [start:lifespan] import os from collections.abc import AsyncIterator from contextlib import asynccontextmanager @@ -44,6 +45,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]: app = FastAPI(lifespan=lifespan) +# --8<-- [end:lifespan] @app.get("/hello/{name}") diff --git a/examples/session_api.py b/examples/session_api.py new file mode 100644 index 0000000..0a49977 --- /dev/null +++ b/examples/session_api.py @@ -0,0 +1,113 @@ +"""Server-side sessions for an API client that keeps its own token. + +``/login`` returns the session token in the response body; the client stores +it and sends it back as ``Authorization: Bearer `` or in the +``X-Session-Token`` header. ``/profile`` needs a logged-in user, ``/public`` +works with or without a session, and ``/logout`` deletes the session so the +token stops working. For browsers, prefer the HttpOnly cookie of +``session_login.py``. + +Run it from a checkout (see ``examples/README.md``):: + + uv run --with "fastapi-cli[standard]" fastapi dev examples/session_api.py +""" + +import os +import secrets +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi import HTTPException +from pydantic import BaseModel + +from fastapi_cachex import BackendProxy +from fastapi_cachex.backends import MemoryBackend +from fastapi_cachex.session import FastAPICacheXSessionMiddleware +from fastapi_cachex.session import SessionConfig +from fastapi_cachex.session import SessionManager +from fastapi_cachex.session import SessionUser +from fastapi_cachex.session.dependencies import AuthenticatedSession +from fastapi_cachex.session.dependencies import OptionalSession +from fastapi_cachex.session.dependencies import SessionDep + +backend = MemoryBackend() +BackendProxy.set(backend) + +config = SessionConfig( + # At least 32 characters. Set a real random value in production, e.g. + # `python -c "import secrets; print(secrets.token_urlsafe(48))"`. + secret_key=os.environ.get( + "SESSION_SECRET_KEY", "dev-only-placeholder-change-me-before-deploying" + ), + session_ttl=3600, # 1 hour +) +session_manager = SessionManager(backend, config) + + +@asynccontextmanager +async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + """Stop the memory backend's cleanup task on shutdown.""" + yield + await backend.aclose() + + +app = FastAPI(lifespan=lifespan) +app.add_middleware( + FastAPICacheXSessionMiddleware, session_manager=session_manager, config=config +) + +# Demo accounts only. Store password hashes (argon2, bcrypt) in a real app. +DEMO_USERS = {"alice": "alice-demo-password"} + + +class Credentials(BaseModel): + """Login form. + + Credentials arrive in the JSON request body, never in the query string, + which ends up in browser history and access logs. + """ + + username: str + password: str + + +@app.post("/login") +async def login(credentials: Credentials) -> dict[str, str]: + """Check the password and return the token of a new session.""" + expected = DEMO_USERS.get(credentials.username) + if expected is None or not secrets.compare_digest(credentials.password, expected): + raise HTTPException(status_code=401, detail="Wrong username or password") + user = SessionUser( + user_id=credentials.username, username=credentials.username, roles=["user"] + ) + _, token = await session_manager.create_session(user=user) + # The client stores the token and sends it on later requests in the + # Authorization or X-Session-Token header. + return {"token": token} + + +@app.get("/profile") +async def get_profile(session: AuthenticatedSession) -> dict[str, object]: + """Requires a session with a user; ``401`` otherwise, anonymous ones included.""" + assert session.user is not None # guaranteed by AuthenticatedSession # noqa: S101 + return { + "user_id": session.user.user_id, + "username": session.user.username, + "roles": session.user.roles, + } + + +@app.get("/public") +async def public_endpoint(session: OptionalSession) -> dict[str, str]: + """Answers with or without a session.""" + if session is not None and session.user is not None: + return {"message": f"Hello, {session.user.username}!"} + return {"message": "Hello, guest!"} + + +@app.post("/logout") +async def logout(session: SessionDep) -> dict[str, bool]: + """Delete the session: its token stops working at once.""" + await session_manager.delete_session(session.session_id) + return {"logged_out": True} diff --git a/examples/session_jwt_claims.py b/examples/session_jwt_claims.py new file mode 100644 index 0000000..a7297a9 --- /dev/null +++ b/examples/session_jwt_claims.py @@ -0,0 +1,196 @@ +"""JWT sessions with custom claims, checked on every request. + +``CustomClaimsJWTSerializer`` issues and verifies the same claims as the +built-in JWT serializer and leaves two hooks for extra ones. +``MultiTenantJWTSerializer`` uses them to put ``tenant_id`` and ``api_version`` +into every token and to reject tokens issued for another tenant. The serializer +is handed to ``SessionManager`` through ``token_serializer``. + +Needs the ``jwt`` extra: ``uv add "fastapi-cachex[jwt]"``. Any backend works; +see ``redis_backend.py`` for Redis. Run it from a checkout (see +``examples/README.md``):: + + uv run --with "fastapi-cli[standard]" fastapi dev examples/session_jwt_claims.py +""" + +# --8<-- [start:serializer] +import os +import secrets +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import datetime +from datetime import timezone +from typing import Any + +import jwt +from fastapi import FastAPI +from fastapi import HTTPException +from pydantic import BaseModel + +from fastapi_cachex.backends import MemoryBackend +from fastapi_cachex.session import FastAPICacheXSessionMiddleware +from fastapi_cachex.session import SessionConfig +from fastapi_cachex.session import SessionManager +from fastapi_cachex.session import SessionUser +from fastapi_cachex.session.dependencies import AuthenticatedSession +from fastapi_cachex.session.models import SessionToken + + +class CustomClaimsJWTSerializer: + """JWT serializer with the built-in claims plus extra ones from subclasses.""" + + # Claims that from_string() requires besides sid, iat and exp. + required_claims: tuple[str, ...] = () + + def __init__(self, config: SessionConfig) -> None: + """Copy the JWT settings from the public ``SessionConfig`` fields.""" + self.secret = config.secret_key.get_secret_value() + self.algorithm = config.jwt_algorithm # must be HS256, HS384 or HS512 + self.issuer = config.jwt_issuer + self.audience = config.jwt_audience + self.leeway = config.jwt_leeway + self.session_ttl = config.session_ttl + + def extra_claims(self, token: SessionToken) -> dict[str, Any]: # noqa: ARG002 + """Return the claims to add to a new token.""" + return {} + + def check_claims(self, payload: dict[str, Any]) -> None: + """Raise ValueError if the extra claims of a verified token are wrong.""" + + def to_string(self, token: SessionToken) -> str: + """Encode a SessionToken as a signed JWT.""" + iat = int(token.issued_at.timestamp()) + if token.expires_at is not None: + exp = int(token.expires_at.timestamp()) + else: + exp = iat + self.session_ttl + + payload: dict[str, Any] = {"sid": token.session_id, "iat": iat, "exp": exp} + if self.issuer: + payload["iss"] = self.issuer + if self.audience: + payload["aud"] = self.audience + payload.update(self.extra_claims(token)) + return jwt.encode(payload, self.secret, algorithm=self.algorithm) + + def from_string(self, token_str: str) -> SessionToken: + """Verify a JWT and turn it back into a SessionToken.""" + try: + payload = jwt.decode( + token_str, + self.secret, + algorithms=[self.algorithm], + issuer=self.issuer, + audience=self.audience, + leeway=self.leeway, + options={"require": ["sid", "iat", "exp", *self.required_claims]}, + ) + except jwt.InvalidTokenError as e: + msg = "Invalid JWT token" + raise ValueError(msg) from e + + self.check_claims(payload) + issued_at = datetime.fromtimestamp(int(payload["iat"]), tz=timezone.utc) + return SessionToken( + session_id=str(payload["sid"]), signature="", issued_at=issued_at + ) + + +# --8<-- [end:serializer] + + +# --8<-- [start:multi-tenant] +class MultiTenantJWTSerializer(CustomClaimsJWTSerializer): + """Adds tenant_id and api_version, and rejects tokens for other tenants.""" + + required_claims = ("tenant_id", "api_version") + + def __init__( + self, config: SessionConfig, tenant_id: str, api_version: str = "v1" + ) -> None: + """Issue and accept tokens for one tenant and API version.""" + super().__init__(config) + self.tenant_id = tenant_id + self.api_version = api_version + + def extra_claims(self, token: SessionToken) -> dict[str, Any]: # noqa: ARG002 + """Put the tenant and API version into every new token.""" + return {"tenant_id": self.tenant_id, "api_version": self.api_version} + + def check_claims(self, payload: dict[str, Any]) -> None: + """Reject a token for another tenant or API version.""" + if payload["tenant_id"] != self.tenant_id: + msg = f"Invalid tenant_id: expected {self.tenant_id}, got {payload['tenant_id']}" + raise ValueError(msg) + if payload["api_version"] != self.api_version: + msg = f"Unsupported API version: {payload['api_version']}" + raise ValueError(msg) + + +# --8<-- [end:multi-tenant] + +# --8<-- [start:setup] +backend = MemoryBackend() +config = SessionConfig( + # HS256 wants a key of at least 32 bytes (HS384: 48, HS512: 64). Set a real + # random value in production, e.g. + # `python -c "import secrets; print(secrets.token_urlsafe(48))"`. + secret_key=os.environ.get( + "SESSION_SECRET_KEY", "dev-only-placeholder-change-me-before-deploying" + ), + token_format="jwt", + jwt_algorithm="HS256", + jwt_issuer="acme-corp", + jwt_audience="acme-api", +) + +# token_serializer replaces the serializer chosen from token_format. +serializer = MultiTenantJWTSerializer(config, tenant_id="acme-corp", api_version="v2") +session_manager = SessionManager(backend, config, token_serializer=serializer) + + +@asynccontextmanager +async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + """Stop the memory backend's cleanup task on shutdown.""" + yield + await backend.aclose() + + +app = FastAPI(lifespan=lifespan) +app.add_middleware( + FastAPICacheXSessionMiddleware, session_manager=session_manager, config=config +) +# --8<-- [end:setup] + +# Demo accounts only. Store password hashes (argon2, bcrypt) in a real app. +DEMO_USERS = {"alice": "alice-demo-password"} + + +class Credentials(BaseModel): + """Login form.""" + + username: str + password: str + + +@app.post("/auth/login") +async def login(credentials: Credentials) -> dict[str, str]: + """Return a JWT that carries the tenant_id and api_version claims.""" + expected = DEMO_USERS.get(credentials.username) + if expected is None or not secrets.compare_digest(credentials.password, expected): + raise HTTPException(status_code=401, detail="Wrong username or password") + user = SessionUser(user_id=credentials.username, username=credentials.username) + _, token = await session_manager.create_session(user=user) + return {"token": token, "token_type": "bearer", "tenant_id": serializer.tenant_id} + + +@app.get("/api/profile") +async def get_profile(session: AuthenticatedSession) -> dict[str, str | None]: + """Protected endpoint; the tenant was checked while decoding the JWT. + + A token for another tenant fails to decode, so the middleware loads no + session and ``AuthenticatedSession`` answers ``401``. + """ + assert session.user is not None # guaranteed by AuthenticatedSession # noqa: S101 + return {"user_id": session.user.user_id, "username": session.user.username} diff --git a/examples/session_redis.py b/examples/session_redis.py new file mode 100644 index 0000000..3157a14 --- /dev/null +++ b/examples/session_redis.py @@ -0,0 +1,171 @@ +"""Server-side sessions on Redis, with bindings, flash messages and logout everywhere. + +The API client gets its token from ``/api/auth/login`` and sends it back as +``Authorization: Bearer `` or in the ``X-Session-Token`` header. The +session is bound to the client's IP address, slides forward while in use, and +carries flash messages between requests. ``/api/auth/logout-all`` ends every +session of the user, on every device. + +Needs the ``redis`` extra (``uv add "fastapi-cachex[redis]"``) and a Redis +server, configured from ``REDIS_HOST``, ``REDIS_PORT``, ``REDIS_DB`` and +``REDIS_PASSWORD`` as in ``redis_backend.py``. Run it from a checkout (see +``examples/README.md``):: + + REDIS_PORT=6379 uv run --with "fastapi-cli[standard]" fastapi dev examples/session_redis.py +""" + +import os +import secrets +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import datetime +from datetime import timezone + +from fastapi import FastAPI +from fastapi import HTTPException +from fastapi import Request +from pydantic import BaseModel + +from fastapi_cachex.backends import AsyncRedisCacheBackend +from fastapi_cachex.session import FastAPICacheXSessionMiddleware +from fastapi_cachex.session import SessionConfig +from fastapi_cachex.session import SessionManager +from fastapi_cachex.session import SessionUser +from fastapi_cachex.session.dependencies import AuthenticatedSession +from fastapi_cachex.session.dependencies import ClientIPDep +from fastapi_cachex.session.dependencies import SessionDep + +# The client connects lazily, so creating it at import needs no server yet. +backend = AsyncRedisCacheBackend( + host=os.environ.get("REDIS_HOST", "127.0.0.1"), + port=int(os.environ.get("REDIS_PORT", "6379")), + db=int(os.environ.get("REDIS_DB", "0")), + password=os.environ.get("REDIS_PASSWORD") or None, + # Namespaces this app's keys on a shared server. + key_prefix="fastapi_cachex_example:", +) + +config = SessionConfig( + # At least 32 characters. Set a real random value in production, e.g. + # `python -c "import secrets; print(secrets.token_urlsafe(48))"`. + secret_key=os.environ.get( + "SESSION_SECRET_KEY", "dev-only-placeholder-change-me-before-deploying" + ), + session_ttl=3600, + sliding_expiration=True, + sliding_threshold=0.5, + ip_binding=True, # reject the token from another IP address + user_agent_binding=False, # optional: reject it from another User-Agent +) +session_manager = SessionManager(backend, config) + + +@asynccontextmanager +async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + """Close the Redis connection pool on shutdown.""" + try: + yield + finally: + await backend.aclose() + + +app = FastAPI(lifespan=lifespan) +app.add_middleware( + FastAPICacheXSessionMiddleware, session_manager=session_manager, config=config +) + +# Demo accounts only. Store password hashes (argon2, bcrypt) in a real app, +# and read the user's ID and roles from your database. +DEMO_USERS = {"alice": "alice-demo-password", "admin": "admin-demo-password"} +DEMO_ROLES = {"alice": ["user"], "admin": ["admin", "user"]} + + +class Credentials(BaseModel): + """Login form.""" + + username: str + password: str + + +@app.post("/api/auth/login") +async def login( + credentials: Credentials, request: Request, client_ip: ClientIPDep +) -> dict[str, object]: + """Check the password and return the token of a new, bound session.""" + expected = DEMO_USERS.get(credentials.username) + if expected is None or not secrets.compare_digest(credentials.password, expected): + raise HTTPException(status_code=401, detail="Wrong username or password") + + user = SessionUser( + user_id=f"user_{credentials.username}", + username=credentials.username, + email=f"{credentials.username}@example.com", + roles=DEMO_ROLES[credentials.username], + ) + # `client_ip` is the address the middleware checks on later requests, + # including behind trusted proxies. + session, token = await session_manager.create_session( + user=user, + ip_address=client_ip, + user_agent=request.headers.get("user-agent"), + ) + + session.add_flash_message("Login successful!", "success") + # Changes to a Session object are saved only by update_session(). + await session_manager.update_session(session) + + return { + "token": token, # the client stores it and sends it on later requests + "user": {"username": user.username, "roles": user.roles}, + } + + +@app.get("/api/user/profile") +async def get_user_profile(session: AuthenticatedSession) -> dict[str, object]: + """Return the user's profile (requires a logged-in user).""" + assert session.user is not None # guaranteed by AuthenticatedSession # noqa: S101 + return { + "user_id": session.user.user_id, + "username": session.user.username, + "email": session.user.email, + "roles": session.user.roles, + "session_created": session.created_at.isoformat(), + "last_accessed": session.last_accessed.isoformat(), + } + + +@app.post("/api/user/update") +async def update_user_profile( + email: str, session: AuthenticatedSession +) -> dict[str, str]: + """Change the email stored in the session.""" + assert session.user is not None # guaranteed by AuthenticatedSession # noqa: S101 + session.user.email = email + session.data["last_updated"] = datetime.now(timezone.utc).isoformat() + await session_manager.update_session(session) + return {"message": "Profile updated"} + + +@app.get("/api/messages") +async def get_flash_messages(session: SessionDep) -> dict[str, object]: + """Return the flash messages once.""" + messages = session.get_flash_messages(clear=True) + # Clearing only changes the in-memory object; save it so the messages + # are not shown again on the next request. + await session_manager.update_session(session) + return {"messages": messages} + + +@app.post("/api/auth/logout") +async def logout(session: SessionDep) -> dict[str, str]: + """End this session; the client should discard its token.""" + await session_manager.delete_session(session.session_id) + return {"message": "Logged out"} + + +@app.post("/api/auth/logout-all") +async def logout_all_devices(session: AuthenticatedSession) -> dict[str, str]: + """End every session of this user, on every device.""" + assert session.user is not None # guaranteed by AuthenticatedSession # noqa: S101 + count = await session_manager.delete_user_sessions(session.user.user_id) + return {"message": f"Logged out from {count} devices"} diff --git a/i18n/zh-TW/GLOSSARY.md b/i18n/zh-TW/GLOSSARY.md index 3a222ee..301c371 100644 --- a/i18n/zh-TW/GLOSSARY.md +++ b/i18n/zh-TW/GLOSSARY.md @@ -7,7 +7,7 @@ - 程式碼、識別字、參數名稱、HTTP 標頭與指令保持原文，並以 inline code 標示： `@cache`、`ttl`、`Cache-Control`、`no-store`、`If-None-Match`。 -- 程式碼區塊內的程式碼不翻譯，只翻譯註解。 +- 程式碼區塊內的程式碼不翻譯，只翻譯註解。以 `--8<--` 從 `examples/` 引入的程式碼例外：註解維持英文，改在區塊前後以中文說明。 - 中文與英文、數字、inline code 之間加一個半形空格；中文句子使用全形標點。 - 下表標示「保留」的詞在中文句子中直接使用英文。 - 章節錨點沿用英文：每個翻譯後的標題都以 `{#id}` 指定與英文頁面相同的錨點，例如 `## 快取鍵 {#cache-keys}`，讓兩種語言的連結可以互換。 diff --git a/i18n/zh-TW/docs/BACKENDS.md b/i18n/zh-TW/docs/BACKENDS.md index d45ee82..b5d4ecb 100644 --- a/i18n/zh-TW/docs/BACKENDS.md +++ b/i18n/zh-TW/docs/BACKENDS.md @@ -142,28 +142,11 @@ BackendProxy.set(backend) 每個後端都有 `aclose()`，用來釋放它持有的連線與背景工作。請在關閉應用程式時，於 FastAPI lifespan 的結尾呼叫它： + ```python -from contextlib import asynccontextmanager - -from fastapi import FastAPI - -from fastapi_cachex import BackendProxy -from fastapi_cachex.backends import AsyncRedisCacheBackend - - -@asynccontextmanager -async def lifespan(app: FastAPI): - backend = AsyncRedisCacheBackend(host="127.0.0.1", port=6379) - BackendProxy.set(backend) - try: - yield - finally: - BackendProxy.set(None) - await backend.aclose() - - -app = FastAPI(lifespan=lifespan) +--8<-- "examples/redis_backend.py:lifespan" ``` + 每種後端都適用同一個 lifespan。後端也是非同步 context manager，因此 `async with MemcachedBackend(servers=[...]) as backend:` 會在區塊結束時關閉它，區塊拋出例外時也一樣。 diff --git a/i18n/zh-TW/docs/HTTP_CACHING.md b/i18n/zh-TW/docs/HTTP_CACHING.md index 3120123..f70f49b 100644 --- a/i18n/zh-TW/docs/HTTP_CACHING.md +++ b/i18n/zh-TW/docs/HTTP_CACHING.md @@ -6,30 +6,13 @@ ## `@cache` 裝飾器 {#the-cache-decorator} + ```python -from fastapi import FastAPI -from fastapi_cachex import cache - -app = FastAPI() - - -@app.get("/") -@cache(ttl=60) # 快取 60 秒 -async def read_root(): - return {"Hello": "World"} - - -@app.get("/no-cache") -@cache(no_cache=True) # 一律重新驗證：每個請求都會執行 handler -async def non_cache_endpoint(): - return {"Hello": "World"} - - -@app.get("/no-store") -@cache(no_store=True) # 任何地方都不儲存這個回應 -async def non_store_endpoint(): - return {"Hello": "World"} +--8<-- "examples/http_cache.py:routes" ``` + + +`ttl=60` 會在 60 秒內提供儲存的回應，`no_cache=True` 讓用戶端每次都重新驗證，`private=True` 則讓回應不存入共用的後端。`no_store=True` 讓回應不存入任何快取；所有選項列在 [Cache-Control 指令](#cache-control-directives)。 只有 GET 請求會被快取；其他方法照常執行 handler。handler 不需要宣告 `Request` 參數：缺少時裝飾器會自動加上。如果尚未設定任何後端，`@cache` 會改用 `MemoryBackend`，並在每個行程記錄一次警告（見[後端](BACKENDS.md#in-memory-default)）。 @@ -457,5 +440,13 @@ add_routes( > > 呼叫 `add_routes()` 時若未傳入 `dependencies`，會發出 `UserWarning`。0.4.0 版將要求必須傳入此參數，並將 `include_content_preview` 預設改為關閉（[#298](https://github.com/allen0099/FastAPI-CacheX/issues/298)）。若本機或測試用的應用程式確實要保持開放，請傳入 `dependencies=[]` 明確選擇不設防護，這樣就不會出現警告。 +可執行範例以環境變數中的權杖保護它們，未設定該變數時路由一律拒絕存取： + + +```python +--8<-- "examples/http_cache.py:admin" +``` + + > [!NOTE] > Memcached 無法列舉鍵，因此在 Memcached 上這兩個路由都不會回傳任何內容。 diff --git a/i18n/zh-TW/docs/JWT_CLAIMS.md b/i18n/zh-TW/docs/JWT_CLAIMS.md index 2822e5a..069e77a 100644 --- a/i18n/zh-TW/docs/JWT_CLAIMS.md +++ b/i18n/zh-TW/docs/JWT_CLAIMS.md @@ -177,77 +177,11 @@ await session_manager.delete_session("session-abc123") 下面的基底類別做的事與內建的 `JWTTokenSerializer` 相同，並為額外的 claim 留下兩個掛鉤。它從 `SessionConfig` 的公開欄位讀取設定並自行保存，而不是存取 `JWTTokenSerializer` 的私有屬性，因為那些屬性在任何版本都可能改變。它與內建序列化器一樣，在 `to_string()` 中採用 `token.expires_at`，讓 `exp` 持續跟著滑動過期。 + ```python -from __future__ import annotations - -from datetime import datetime, timezone -from typing import Any - -import jwt - -from fastapi_cachex.session import SessionConfig -from fastapi_cachex.session.models import SessionToken - - -class CustomClaimsJWTSerializer: - """JWT serializer with the built-in claims plus extra ones from subclasses.""" - - # 除了 sid、iat 與 exp 之外，from_string() 還要求的 claim。 - required_claims: tuple[str, ...] = () - - def __init__(self, config: SessionConfig) -> None: - self.secret = config.secret_key.get_secret_value() - self.algorithm = config.jwt_algorithm # 必須是 HS256、HS384 或 HS512 - self.issuer = config.jwt_issuer - self.audience = config.jwt_audience - self.leeway = config.jwt_leeway - self.session_ttl = config.session_ttl - - def extra_claims(self, token: SessionToken) -> dict[str, Any]: - """Return the claims to add to a new token.""" - return {} - - def check_claims(self, payload: dict[str, Any]) -> None: - """Raise ValueError if the extra claims of a verified token are wrong.""" - - def to_string(self, token: SessionToken) -> str: - """Encode a SessionToken as a signed JWT.""" - iat = int(token.issued_at.timestamp()) - if token.expires_at is not None: - exp = int(token.expires_at.timestamp()) - else: - exp = iat + self.session_ttl - - payload: dict[str, Any] = {"sid": token.session_id, "iat": iat, "exp": exp} - if self.issuer: - payload["iss"] = self.issuer - if self.audience: - payload["aud"] = self.audience - payload.update(self.extra_claims(token)) - return jwt.encode(payload, self.secret, algorithm=self.algorithm) - - def from_string(self, token_str: str) -> SessionToken: - """Verify a JWT and turn it back into a SessionToken.""" - try: - payload = jwt.decode( - token_str, - self.secret, - algorithms=[self.algorithm], - issuer=self.issuer, - audience=self.audience, - leeway=self.leeway, - options={"require": ["sid", "iat", "exp", *self.required_claims]}, - ) - except jwt.InvalidTokenError as e: - msg = "Invalid JWT token" - raise ValueError(msg) from e - - self.check_claims(payload) - issued_at = datetime.fromtimestamp(int(payload["iat"]), tz=timezone.utc) - return SessionToken( - session_id=str(payload["sid"]), signature="", issued_at=issued_at - ) +--8<-- "examples/session_jwt_claims.py:serializer" ``` + PyJWT 預設會驗證簽章、`exp`、`iat` 與（存在時的）`nbf`，並在傳入 `issuer`／`audience` 時驗證 `iss`／`aud`。內建序列化器的兩項檢查在這裡沒有重複：它會拒絕非對稱的 `jwt_algorithm`，並在 `secret_key` 短於 HMAC 輸出長度時發出警告。這個類別同樣以 `secret_key` 簽署，因此請使用 `HS*` 演算法；若要使用非對稱演算法，請在類別中保存私鑰與公鑰，並在 `jwt.encode()` 與 `jwt.decode()` 中使用它們。 @@ -274,83 +208,26 @@ class ExtendedJWTSerializer(CustomClaimsJWTSerializer): ### 範例 2：加入多租戶的自訂 claim {#example-2-adding-multi-tenant-custom-claims} + ```python -from typing import Any - -from fastapi_cachex.session import SessionConfig -from fastapi_cachex.session.models import SessionToken - - -class MultiTenantJWTSerializer(CustomClaimsJWTSerializer): - """Adds tenant_id and api_version, and rejects tokens for other tenants.""" - - required_claims = ("tenant_id", "api_version") - - def __init__( - self, config: SessionConfig, tenant_id: str, api_version: str = "v1" - ) -> None: - super().__init__(config) - self.tenant_id = tenant_id - self.api_version = api_version - - def extra_claims(self, token: SessionToken) -> dict[str, Any]: - return {"tenant_id": self.tenant_id, "api_version": self.api_version} - - def check_claims(self, payload: dict[str, Any]) -> None: - if payload["tenant_id"] != self.tenant_id: - msg = f"Invalid tenant_id: expected {self.tenant_id}, got {payload['tenant_id']}" - raise ValueError(msg) - if payload["api_version"] != self.api_version: - msg = f"Unsupported API version: {payload['api_version']}" - raise ValueError(msg) +--8<-- "examples/session_jwt_claims.py:multi-tenant" ``` + ### 使用自訂序列化器 {#using-a-custom-serializer} #### 做法 1：傳給 `SessionManager`（建議） {#option-1-pass-it-to-sessionmanager-recommended} + ```python -from fastapi import FastAPI - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, -) - -app = FastAPI() - -# 建立後端與設定 -backend = AsyncRedisCacheBackend(host="localhost", port=6379) -config = SessionConfig( - secret_key="your-secret-key-at-least-32-characters", - token_format="jwt", - jwt_algorithm="HS256", - jwt_issuer="your-company", - jwt_audience="your-api", -) - -# 建立自訂序列化器 -custom_serializer = MultiTenantJWTSerializer( - config=config, - tenant_id="acme-corp", - api_version="v2", -) - -# 初始化 SessionManager -manager = SessionManager(backend, config, token_serializer=custom_serializer) - -# 加入中介軟體 -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=manager, - config=config, -) +--8<-- "examples/session_jwt_claims.py:setup" ``` + 提供 `token_serializer` 時，它會取代依 `token_format` 選擇的內建序列化器。但仍請保留 `token_format="jwt"`：若設為 `"simple"`，`SessionManager` 會對解析出的權杖額外執行自己的 HMAC 簽章檢查，而以 JWT 為基礎的序列化器產生的權杖通不過這項檢查。 +範例使用 `MemoryBackend`，因此不需要伺服器即可執行；任何後端都可以，例如[後端](BACKENDS.md#closing-a-backend)中的 Redis 設定。 + #### 做法 2：繼承 `SessionManager`（進階） {#option-2-subclass-sessionmanager-advanced} ```python @@ -381,89 +258,13 @@ manager = MultiTenantSessionManager(backend, config, tenant_id="acme-corp") ## 完整應用程式範例 {#complete-application-example} -```python -from __future__ import annotations - -from fastapi import Depends, FastAPI, HTTPException -from pydantic import BaseModel - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - Session, - SessionConfig, - SessionManager, - SessionUser, - require_user_session, -) - -# 使用上面定義的 MultiTenantJWTSerializer +[`examples/session_jwt_claims.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_jwt_claims.py) 把上面的各個部分組合成可執行的應用程式（程式碼註解為英文）： -app = FastAPI() - -# 初始化 -backend = AsyncRedisCacheBackend(host="localhost", port=6379) -config = SessionConfig( - secret_key="your-secret-key-min-32-chars-long!!", - token_format="jwt", - jwt_algorithm="HS256", - jwt_issuer="acme-corp", - jwt_audience="acme-api", -) - -# 建立自訂序列化器 -serializer = MultiTenantJWTSerializer( - config=config, - tenant_id="acme-corp", - api_version="v2", -) - -manager = SessionManager(backend, config, token_serializer=serializer) - -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=manager, - config=config, -) - - -class LoginRequest(BaseModel): - username: str - password: str - - -@app.post("/auth/login") -async def login(credentials: LoginRequest) -> dict[str, str]: - """Login endpoint that returns a JWT containing tenant_id.""" - # 驗證使用者（省略） - if credentials.username != "admin": - raise HTTPException(status_code=401, detail="Invalid credentials") - - user = SessionUser(user_id="123", username=credentials.username) - session, token = await manager.create_session(user=user) - - # 權杖現在包含 tenant_id 與 api_version claim - return { - "token": token, - "token_type": "bearer", - "tenant_id": "acme-corp", # 也可以從設定讀取 - } - - -@app.get("/api/profile") -async def get_profile( - session: Session = Depends(require_user_session), -) -> dict[str, str | None]: - """Protected endpoint; tenant_id is validated automatically.""" - # tenant_id 與 api_version 已在解碼 JWT 時驗證過。 - # 其他租戶的權杖會解碼失敗，因此中介軟體不會載入 - # Session，require_user_session 會回應 401。 - assert session.user is not None - return { - "user_id": session.user.user_id, - "username": session.user.username, - } + +```python +--8<-- "examples/session_jwt_claims.py" ``` + ## 安全性考量 {#security-considerations} diff --git a/i18n/zh-TW/docs/SESSION.md b/i18n/zh-TW/docs/SESSION.md index 93c1bfc..ccf4fdb 100644 --- a/i18n/zh-TW/docs/SESSION.md +++ b/i18n/zh-TW/docs/SESSION.md @@ -48,110 +48,28 @@ uv add "fastapi-cachex[jwt]" ### 2. 基本用法 {#2-basic-usage} -```python -from fastapi import Depends, FastAPI, HTTPException -from pydantic import BaseModel - -from fastapi_cachex.backends import MemoryBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, - SessionUser, - get_optional_session, - get_session, -) -from fastapi_cachex.session.dependencies import AuthenticatedSession - -# 建立 FastAPI 應用程式 -app = FastAPI() - -# Session 設定（API 優先架構：由用戶端管理權杖） -config = SessionConfig( - secret_key="your-secret-key-min-32-chars-long!!!", # 至少 32 個字元 - session_ttl=3600, # 1 小時 -) - -# 設定後端與 Session 管理器 -backend = MemoryBackend() -session_manager = SessionManager(backend, config) - -# 加入 Session 中介軟體（SessionMiddleware 已棄用，將於 0.4.0 移除） -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=session_manager, - config=config, -) - -# 或者，也可以將管理器註冊到 proxy，而不是直接傳入： -# -# from fastapi_cachex.session import SessionManagerProxy -# -# SessionManagerProxy.set(session_manager) -# app.add_middleware(FastAPICacheXSessionMiddleware) # 從 proxy 取得 -# -# 省略 `config` 時，中介軟體會使用 `session_manager.config`。 +以下是 [`examples/session_api.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_api.py)（程式碼註解為英文）：API 用戶端登入後保存取得的權杖，並在之後的請求中送出它。 + +```python +--8<-- "examples/session_api.py" +``` + -# 帳號密碼放在 JSON 請求本文中，絕不放在查詢字串， -# 查詢字串會留在瀏覽器歷史紀錄與存取日誌裡。 -class LoginRequest(BaseModel): - username: str - password: str +也可以不把管理器傳給中介軟體，而是將它註冊到 proxy。省略 `config` 時，中介軟體會使用 `session_manager.config`： +```python +from fastapi_cachex.session import SessionManagerProxy -# 登入端點 -@app.post("/login") -async def login(credentials: LoginRequest): - # 驗證使用者（此處為簡化版） - if credentials.username != "admin" or credentials.password != "secret": - raise HTTPException(status_code=401, detail="Invalid credentials") - - # 建立 Session - user = SessionUser( - user_id="123", - username=credentials.username, - roles=["admin"], - ) - session, token = await session_manager.create_session(user=user) - - # 回傳權杖供用戶端保存（localStorage/sessionStorage）。 - # 用戶端之後的請求會在 Authorization 或 X-Session-Token 標頭中送出它。 - return {"message": "Login successful", "token": token} - - -# 需要已登入使用者的端點 -@app.get("/profile") -async def get_profile(session: AuthenticatedSession): - """Requires a session with a user; 401 otherwise, anonymous sessions included.""" - return { - "user_id": session.user.user_id, - "username": session.user.username, - "roles": session.user.roles, - } - - -# 可選驗證的端點 -@app.get("/public") -async def public_endpoint(session=Depends(get_optional_session)): - """Accessible with or without a session.""" - if session and session.user: - return {"message": f"Hello, {session.user.username}!"} - return {"message": "Hello, guest!"} - - -# 登出端點 -@app.post("/logout") -async def logout(session=Depends(get_session)): - await session_manager.delete_session(session.session_id) - return {"message": "Logged out"} +SessionManagerProxy.set(session_manager) +app.add_middleware(FastAPICacheXSessionMiddleware) # 從 proxy 取得 ``` 當請求沒有帶著有效的 Session 時，`get_session`（及其別名 `require_session`）會拋出 `401 Authentication required`，並附上 `WWW-Authenticate: Bearer` 標頭。格式錯誤、偽造、已過期、已失效或未通過綁定檢查的權杖，在中介軟體層級都不會被視為錯誤：請求只會在沒有 Session 的情況下繼續處理。 依賴項回傳的 Session 物件是後端的 `Session` 模型。由 `FastAPICacheXSessionMiddleware` 從 `request.session` 建立的 Session（見下方的遷移一節）是匿名的，因此 `session.user` 為 `None`。 -`get_session` 也接受這種 Session，因此它只能證明請求帶著「某個」Session，而不能證明有人登入。任何訪客只要進入會寫入 `request.session` 的路由（購物車、CSRF 值），就會得到一個。需要已登入使用者的路由，請改用 `require_user_session`（或其型別註記形式 `AuthenticatedSession`）保護，它在 `session.user` 為 `None` 時同樣回應 `401`，上方的 `/profile` 就是這樣做的。`/logout` 只會刪除 Session，因此使用 `get_session` 就足夠。 +`get_session` 也接受這種 Session，因此它只能證明請求帶著「某個」Session，而不能證明有人登入。任何訪客只要進入會寫入 `request.session` 的路由（購物車、CSRF 值），就會得到一個。需要已登入使用者的路由，請改用 `require_user_session`（或其型別註記形式 `AuthenticatedSession`）保護，它在 `session.user` 為 `None` 時同樣回應 `401`，上方的 `/profile` 就是這樣做的。`/logout` 只會刪除 Session，因此使用 `SessionDep`（`get_session` 的型別註記形式）就足夠；`/public` 則使用 `OptionalSession`（`get_optional_session`），沒有 Session 時得到 `None`，而不是回應 `401`。 `UserSessionDep` 雖然名稱如此，卻不會檢查使用者；在 0.4.0 之前它是 `SessionDep` 的別名，0.4.0 預計改為要求使用者。 @@ -159,164 +77,13 @@ async def logout(session=Depends(get_session)): ### 3. 完整範例（Redis 後端） {#3-full-example-redis-backend} -```python -from datetime import datetime, timezone - -from fastapi import Depends, FastAPI, HTTPException, Request -from pydantic import BaseModel - -from fastapi_cachex.backends import AsyncRedisCacheBackend -from fastapi_cachex.session import ( - FastAPICacheXSessionMiddleware, - SessionConfig, - SessionManager, - SessionUser, - get_session, -) -from fastapi_cachex.session.dependencies import AuthenticatedSession, ClientIPDep - -app = FastAPI() - -# Redis 後端 -backend = AsyncRedisCacheBackend( - host="localhost", - port=6379, - db=0, -) - -# 包含安全性選項的 Session 設定 -config = SessionConfig( - secret_key="your-very-secret-key-at-least-32-characters-long!!", - session_ttl=3600, - sliding_expiration=True, - sliding_threshold=0.5, - ip_binding=True, # 啟用 IP 綁定 - user_agent_binding=False, # UA 綁定（可選） -) - -session_manager = SessionManager(backend, config) - -app.add_middleware( - FastAPICacheXSessionMiddleware, - session_manager=session_manager, - config=config, -) - - -class LoginRequest(BaseModel): - username: str - password: str - - -@app.post("/api/auth/login") -async def login(credentials: LoginRequest, request: Request, client_ip: ClientIPDep): - # 驗證使用者（實際上應查詢資料庫） - username = credentials.username - if not authenticate_user(username, credentials.password): - raise HTTPException(status_code=401, detail="Invalid credentials") - - # 建立 Session - user = SessionUser( - user_id=get_user_id(username), - username=username, - email=f"{username}@example.com", - roles=get_user_roles(username), - ) - - # 收集綁定所需的用戶端資訊。`client_ip` 就是中介軟體之後 - # 會檢查的位址，在受信任的 proxy 後方也是如此。 - user_agent = request.headers.get("user-agent") - - session, token = await session_manager.create_session( - user=user, - ip_address=client_ip, - user_agent=user_agent, - ) - - # 加入 Flash 訊息 - session.add_flash_message("Login successful!", "success") - await session_manager.update_session(session) - - return { - "message": "Login successful", - "token": token, # 用戶端保存這個權杖，並在之後的請求中送出 - "user": { - "username": user.username, - "roles": user.roles, - }, - } - +以下是 [`examples/session_redis.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/session_redis.py)（程式碼註解為英文）。它與 [`examples/redis_backend.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/redis_backend.py) 一樣，從 `REDIS_HOST`、`REDIS_PORT`、`REDIS_DB` 與 `REDIS_PASSWORD` 讀取 Redis 設定。 -@app.get("/api/user/profile") -async def get_user_profile(session: AuthenticatedSession): - """Return the user's profile (requires a logged-in user).""" - return { - "user_id": session.user.user_id, - "username": session.user.username, - "email": session.user.email, - "roles": session.user.roles, - "session_created": session.created_at.isoformat(), - "last_accessed": session.last_accessed.isoformat(), - } - - -@app.post("/api/user/update") -async def update_user_profile( - email: str, - session: AuthenticatedSession, -): - """Update the user's profile.""" - session.user.email = email - session.data["last_updated"] = datetime.now(timezone.utc).isoformat() - - # 儲存更新後的 Session - await session_manager.update_session(session) - - return {"message": "Profile updated"} - - -@app.get("/api/messages") -async def get_flash_messages(session=Depends(get_session)): - """Return and clear the flash messages.""" - messages = session.get_flash_messages(clear=True) - # 清除只會改變記憶體中的物件；請儲存它， - # 以免下一個請求再次顯示這些訊息。 - await session_manager.update_session(session) - return {"messages": messages} - - -@app.post("/api/auth/logout") -async def logout(session=Depends(get_session)): - """Log out.""" - await session_manager.delete_session(session.session_id) - - # 用戶端應丟棄已保存的權杖 - return {"message": "Logged out successfully"} - - -@app.post("/api/auth/logout-all") -async def logout_all_devices(session: AuthenticatedSession): - """Log out from all devices.""" - user_id = session.user.user_id - count = await session_manager.delete_user_sessions(user_id) - return {"message": f"Logged out from {count} devices"} - - -# 輔助函式（僅為示意） -def authenticate_user(username: str, password: str) -> bool: - # 實際的實作會查詢資料庫並驗證密碼雜湊 - return True - - -def get_user_id(username: str) -> str: - # 實際的實作會從資料庫讀取 - return f"user_{username}" - - -def get_user_roles(username: str) -> list[str]: - # 實際的實作會從資料庫讀取 - return ["user"] if username != "admin" else ["admin", "user"] + +```python +--8<-- "examples/session_redis.py" ``` + 在 handler 中對 `Session` 物件所做的變更（Flash 訊息、`session.data`、`session.user`），只有在呼叫 `session_manager.update_session(session)` 時才會被儲存。 diff --git a/i18n/zh-TW/docs/STATE.md b/i18n/zh-TW/docs/STATE.md index ccf10c9..2969a9e 100644 --- a/i18n/zh-TW/docs/STATE.md +++ b/i18n/zh-TW/docs/STATE.md @@ -8,55 +8,15 @@ State 與 HTTP 快取存放在同一個後端，但使用自己的鍵前綴（ 本指南中的所有內容也都可以從頂層的 `fastapi_cachex` 套件匯入。 -完整可執行範例（英文）：[`examples/oauth_state.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/oauth_state.py)。 +下方的快速開始就是完整可執行的範例 [`examples/oauth_state.py`](https://github.com/allen0099/FastAPI-CacheX/blob/master/examples/oauth_state.py)，程式碼註解為英文。 ## 快速開始 {#quick-start} + ```python -import secrets - -from fastapi import FastAPI, HTTPException, Request -from fastapi.responses import RedirectResponse - -from fastapi_cachex import BackendProxy -from fastapi_cachex.backends import MemoryBackend -from fastapi_cachex.state import StateError, StateManagerDep - -app = FastAPI() -BackendProxy.set(MemoryBackend()) - -BINDING_COOKIE = "oauth_binding" - - -@app.get("/login") -async def login(states: StateManagerDep): - nonce = secrets.token_urlsafe(32) - state = await states.create_state(binding=nonce, metadata={"next": "/dashboard"}) - response = RedirectResponse( - f"https://provider.example.com/authorize?state={state}&client_id=..." - ) - # 使用 Lax 而非 Strict：回呼是從提供者發起的跨站導覽。 - response.set_cookie( - BINDING_COOKIE, nonce, max_age=600, httponly=True, secure=True, samesite="lax" - ) - return response - - -@app.get("/callback") -async def callback(request: Request, state: str, code: str, states: StateManagerDep): - try: - # 一次性：取出時即刪除。除非是這個瀏覽器發起的流程，否則會被拒絕。 - data = await states.consume_state( - state, binding=request.cookies.get(BINDING_COOKIE) - ) - except StateError as e: # 未知、已過期、格式錯誤或發給其他瀏覽器的 state - raise HTTPException(status_code=400, detail="Invalid state") from e - - # 以 code 換取權杖、建立 Session…… - response = RedirectResponse(data.metadata.get("next", "/")) - response.delete_cookie(BINDING_COOKIE) - return response +--8<-- "examples/oauth_state.py" ``` + 若提供者以 POST 送出回呼（`response_mode=form_post`），`SameSite=Lax` 的 Cookie 不會隨這個跨站 POST 送出；此時綁定用的 Cookie 請改用 `samesite="none"`（需要 `secure=True`）。在另一個分頁再次開始登入會覆寫這個 Cookie，因此第一個分頁的回呼會被拒絕；使用者只要重新登入即可。 diff --git a/tests/test_examples.py b/tests/test_examples.py index 930bd21..24bd633 100644 --- a/tests/test_examples.py +++ b/tests/test_examples.py @@ -5,10 +5,10 @@ `TestClient`. A `DeprecationWarning` fails the test, so an example cannot keep showing an API we are phasing out. -`session_jwt` and `redis_backend` need optional packages and are skipped -without them (checked with `find_spec`, never imported here). `redis_backend` -also talks to a real server, so it follows the opt-in rules in -`tests/live_servers.py`. +`session_jwt`, `session_jwt_claims`, `redis_backend` and `session_redis` need +optional packages and are skipped without them (checked with `find_spec`, never +imported here). `redis_backend` and `session_redis` also talk to a real server, +so they follow the opt-in rules in `tests/live_servers.py`. """ import importlib.util @@ -26,6 +26,7 @@ from fastapi_cachex.exceptions import BackendNotFoundError from fastapi_cachex.manager_proxy import CacheManagerProxy from fastapi_cachex.proxy import BackendProxy +from fastapi_cachex.session.exceptions import SessionSecurityError from fastapi_cachex.session.proxy import SessionManagerProxy from fastapi_cachex.state.proxy import StateManagerProxy from tests.live_servers import REDIS_HOST @@ -74,8 +75,11 @@ def test_every_example_has_a_test() -> None: "oauth_state", "rate_limit", "redis_backend", + "session_api", "session_jwt", + "session_jwt_claims", "session_login", + "session_redis", } assert {p.stem for p in EXAMPLES_DIR.glob("*.py")} == tested readme = (EXAMPLES_DIR / "README.md").read_text(encoding="utf-8") @@ -301,6 +305,66 @@ def test_session_jwt() -> None: assert client.get("/me", headers=auth).status_code == 401 +def test_session_api() -> None: + example = load_example("session_api") + credentials = {"username": "alice", "password": "alice-demo-password"} + with TestClient(example.app) as client: + assert client.get("/public").json() == {"message": "Hello, guest!"} + assert client.get("/profile").status_code == 401 + wrong = {"username": "alice", "password": "nope"} + assert client.post("/login", json=wrong).status_code == 401 + + token = client.post("/login", json=credentials).json()["token"] + # The token is only in the body: no cookie for an API client. + assert not client.cookies.get("session") + bearer = {"Authorization": f"Bearer {token}"} + header = {"X-Session-Token": token} + assert client.get("/profile", headers=bearer).json() == { + "user_id": "alice", + "username": "alice", + "roles": ["user"], + } + assert client.get("/public", headers=header).json() == { + "message": "Hello, alice!" + } + + assert client.post("/logout", headers=bearer).status_code == 200 + assert client.get("/profile", headers=bearer).status_code == 401 + assert client.post("/logout", headers=bearer).status_code == 401 + + +@pytest.mark.skipif( + importlib.util.find_spec("jwt") is None, + reason="session_jwt_claims needs the jwt extra (PyJWT)", +) +def test_session_jwt_claims() -> None: + example = load_example("session_jwt_claims") + credentials = {"username": "alice", "password": "alice-demo-password"} + with TestClient(example.app) as client: + wrong = {"username": "alice", "password": "nope"} + assert client.post("/auth/login", json=wrong).status_code == 401 + issued = client.post("/auth/login", json=credentials) + assert issued.status_code == 200 + token = issued.json()["token"] + claims = example.jwt.decode(token, options={"verify_signature": False}) + assert claims["tenant_id"] == "acme-corp" + assert claims["api_version"] == "v2" + assert claims["iss"] == "acme-corp" + auth = {"Authorization": f"Bearer {token}"} + assert client.get("/api/profile", headers=auth).json() == { + "user_id": "alice", + "username": "alice", + } + + # The same session, signed with the same key, for another tenant. + other = example.MultiTenantJWTSerializer(example.config, tenant_id="other") + foreign = other.to_string(example.serializer.from_string(token)) + with pytest.raises(ValueError, match="Invalid tenant_id"): + example.serializer.from_string(foreign) + foreign_auth = {"Authorization": f"Bearer {foreign}"} + assert client.get("/api/profile", headers=foreign_auth).status_code == 401 + + def test_oauth_state() -> None: example = load_example("oauth_state") # https: the binding cookie is Secure. @@ -416,3 +480,68 @@ def test_redis_backend(monkeypatch: pytest.MonkeyPatch) -> None: # The lifespan unregistered the backend on shutdown. with pytest.raises(BackendNotFoundError): BackendProxy.get() + + +@pytest.mark.skipif( + importlib.util.find_spec("redis") is None, + reason="session_redis needs the redis extra", +) +def test_session_redis(monkeypatch: pytest.MonkeyPatch) -> None: + reason = redis_skip_reason() + if reason is not None: + pytest.skip(reason) + monkeypatch.setenv("REDIS_HOST", REDIS_HOST) + monkeypatch.setenv("REDIS_PORT", str(REDIS_PORT)) + monkeypatch.delenv("REDIS_PASSWORD", raising=False) + monkeypatch.delenv("REDIS_DB", raising=False) + example = load_example("session_redis") + credentials = {"username": "alice", "password": "alice-demo-password"} + with TestClient(example.app) as client: + portal = client.portal + assert portal is not None + # clear() removes only this app's namespace, not the whole server. + portal.call(example.backend.clear) + try: + wrong = {"username": "alice", "password": "nope"} + assert client.post("/api/auth/login", json=wrong).status_code == 401 + + first = client.post("/api/auth/login", json=credentials).json() + assert first["user"] == {"username": "alice", "roles": ["user"]} + auth = {"Authorization": f"Bearer {first['token']}"} + + profile = client.get("/api/user/profile", headers=auth).json() + assert profile["user_id"] == "user_alice" + assert profile["email"] == "alice@example.com" + + # The flash message from the login is shown once. + messages = client.get("/api/messages", headers=auth).json()["messages"] + assert [m["message"] for m in messages] == ["Login successful!"] + assert client.get("/api/messages", headers=auth).json() == {"messages": []} + + updated = client.post( + "/api/user/update", params={"email": "a@example.org"}, headers=auth + ) + assert updated.status_code == 200 + profile = client.get("/api/user/profile", headers=auth).json() + assert profile["email"] == "a@example.org" + + # The session is bound to the client's IP address. (A second + # TestClient would run on another event loop than the Redis pool.) + with pytest.raises(SessionSecurityError): + portal.call( + example.session_manager.get_session, first["token"], "203.0.113.9" + ) + + second = client.post("/api/auth/login", json=credentials).json() + auth2 = {"X-Session-Token": second["token"]} + assert client.post("/api/auth/logout", headers=auth2).status_code == 200 + assert client.get("/api/user/profile", headers=auth2).status_code == 401 + + third = client.post("/api/auth/login", json=credentials).json() + auth3 = {"Authorization": f"Bearer {third['token']}"} + ended = client.post("/api/auth/logout-all", headers=auth3).json() + assert ended == {"message": "Logged out from 2 devices"} + assert client.get("/api/user/profile", headers=auth).status_code == 401 + assert client.get("/api/user/profile", headers=auth3).status_code == 401 + finally: + portal.call(example.backend.clear)