d3vyce/fastapi-toolsets
1
0
Fork 0
mirror of https://github.com/d3vyce/fastapi-toolsets.git synced 2026-08-06 08:34:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: rename nonce by state_token

Browse Source
This commit is contained in:
d3vyce
2026-05-07 19:01:59 -04:00
parent aeb9e0d9b7
commit 3c2d14b70a
5 changed files with 70 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/module/security.md
+16 -25
View File
@@ -174,15 +174,6 @@ async def profile(user: User = Security(bearer.require(role=Role.USER))):
The `prefix` (for `BearerTokenAuth`), cookie name and `secret_key` (for
`CookieAuth`), and header name (for `APIKeyHeaderAuth`) are always preserved.
`.require()` instances work transparently inside `MultiAuth`:
```python
multi = MultiAuth(
user_bearer.require(role=Role.USER),
org_bearer.require(role=Role.ADMIN),
)
```
## MultiAuth
[`MultiAuth`](../reference/security.md#fastapi_toolsets.security.MultiAuth) combines multiple auth sources into a single callable. Sources are tried in order; the first one that finds a credential wins.
@@ -285,24 +276,24 @@ Returns a `(authorization_url, token_url, userinfo_url)` tuple. `userinfo_url` i
### Authorization redirect
[`oauth_build_authorization_redirect()`](../reference/security.md#fastapi_toolsets.security.oauth_build_authorization_redirect) constructs the redirect to the provider's authorization page. It requires a `nonce` — a random CSRF token generated by [`oauth_generate_nonce()`](../reference/security.md#fastapi_toolsets.security.oauth_generate_nonce) — that must be stored server-side (e.g. in the session) and verified on the callback to prevent login-CSRF attacks (RFC 6749 §10.12):
[`oauth_build_authorization_redirect()`](../reference/security.md#fastapi_toolsets.security.oauth_build_authorization_redirect) constructs the redirect to the provider's authorization page. It requires a `state_token` — a random CSRF token generated by [`oauth_generate_state_token()`](../reference/security.md#fastapi_toolsets.security.oauth_generate_state_token) — that must be stored server-side (e.g. in the session) and verified on the callback to prevent login-CSRF attacks ([RFC 6749 §10.12](https://datatracker.ietf.org/doc/html/rfc6749#section-10.12)):
```python
from fastapi import Request
from fastapi_toolsets.security import oauth_build_authorization_redirect, oauth_generate_nonce
from fastapi_toolsets.security import oauth_build_authorization_redirect, oauth_generate_state_token
@app.get("/auth/google/login")
async def google_login(request: Request):
auth_url, _, _ = await oauth_resolve_provider_urls(GOOGLE_DISCOVERY_URL)
nonce = oauth_generate_nonce()
request.session["oauth_nonce"] = nonce # requires SessionMiddleware
state_token = oauth_generate_state_token()
request.session["oauth_state"] = state_token # requires SessionMiddleware
return oauth_build_authorization_redirect(
auth_url,
client_id=GOOGLE_CLIENT_ID,
scopes="openid email profile",
redirect_uri="https://myapp.com/auth/google/callback",
destination="/dashboard",
nonce=nonce,
state_token=state_token,
)
```
@@ -310,7 +301,7 @@ async def google_login(request: Request):
[`oauth_fetch_userinfo()`](../reference/security.md#fastapi_toolsets.security.oauth_fetch_userinfo) performs the two-step exchange: it POSTs the authorization code to the token endpoint, then GETs the userinfo endpoint with the resulting access token.
On the callback, retrieve the stored nonce and pass it to [`oauth_decode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_decode_state) to verify the CSRF token before processing the code:
On the callback, retrieve the stored token and pass it to [`oauth_decode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_decode_state) to verify the CSRF token before processing the code:
```python
from fastapi import HTTPException, Request
@@ -318,11 +309,11 @@ from fastapi_toolsets.security import oauth_decode_state, oauth_fetch_userinfo
@app.get("/auth/google/callback")
async def google_callback(request: Request, code: str, state: str):
# Pop nonce first — single-use, regardless of whether verification succeeds
nonce = request.session.pop("oauth_nonce", None)
if nonce is None:
# Pop token first — single-use, regardless of whether verification succeeds
state_token = request.session.pop("oauth_state", None)
if state_token is None:
raise HTTPException(status_code=400, detail="missing OAuth state")
destination = oauth_decode_state(state, expected_nonce=nonce, fallback="/")
destination = oauth_decode_state(state, expected_state_token=state_token, fallback="/")
if not destination.startswith("/"): # reject absolute URLs to prevent open-redirect
destination = "/"
@@ -346,16 +337,16 @@ Pass `required_scopes` to guard against providers silently granting fewer scopes
### State encoding
[`oauth_encode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_encode_state) and [`oauth_decode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_decode_state) encode and decode the destination URL together with the CSRF nonce embedded in the OAuth `state` parameter. `oauth_decode_state` returns `fallback` if `state` is absent, malformed, or the nonce does not match:
[`oauth_encode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_encode_state) and [`oauth_decode_state()`](../reference/security.md#fastapi_toolsets.security.oauth_decode_state) encode and decode the destination URL together with the CSRF token embedded in the OAuth `state` parameter. `oauth_decode_state` returns `fallback` if `state` is absent, malformed, or the token does not match:
```python
from fastapi_toolsets.security import oauth_encode_state, oauth_decode_state
nonce = "my-random-nonce"
encoded = oauth_encode_state("/dashboard", nonce)
decoded = oauth_decode_state(encoded, expected_nonce=nonce, fallback="/") # "/dashboard"
decoded = oauth_decode_state(encoded, expected_nonce="wrong", fallback="/") # "/"
decoded = oauth_decode_state(None, expected_nonce=nonce, fallback="/") # "/"
state_token = oauth_generate_state_token()
encoded = oauth_encode_state("/dashboard", state_token)
decoded = oauth_decode_state(encoded, expected_state_token=state_token, fallback="/") # "/dashboard"
decoded = oauth_decode_state(encoded, expected_state_token="wrong", fallback="/") # "/"
decoded = oauth_decode_state(None, expected_state_token=state_token, fallback="/") # "/"
```
---
docs/reference/security.md
+2 -2
View File
@@ -15,7 +15,7 @@ from fastapi_toolsets.security import (
oauth_decode_state,
oauth_encode_state,
oauth_fetch_userinfo,
oauth_generate_nonce,
oauth_generate_state_token,
oauth_resolve_provider_urls,
)
```
@@ -34,7 +34,7 @@ from fastapi_toolsets.security import (
## ::: fastapi_toolsets.security.oauth_fetch_userinfo
## ::: fastapi_toolsets.security.oauth_generate_nonce
## ::: fastapi_toolsets.security.oauth_generate_state_token
## ::: fastapi_toolsets.security.oauth_build_authorization_redirect
src/fastapi_toolsets/security/__init__.py
+2 -2
View File
@@ -6,7 +6,7 @@ from .oauth import (
oauth_decode_state,
oauth_encode_state,
oauth_fetch_userinfo,
oauth_generate_nonce,
oauth_generate_state_token,
oauth_resolve_provider_urls,
)
from .sources import APIKeyHeaderAuth, BearerTokenAuth, CookieAuth, MultiAuth
@@ -21,6 +21,6 @@ __all__ = [
"oauth_decode_state",
"oauth_encode_state",
"oauth_fetch_userinfo",
"oauth_generate_nonce",
"oauth_generate_state_token",
"oauth_resolve_provider_urls",
]
src/fastapi_toolsets/security/oauth.py
+19 -22
View File
@@ -103,13 +103,8 @@ async def oauth_fetch_userinfo(
return userinfo_resp.json()
def oauth_generate_nonce() -> str:
"""Generate a cryptographically random nonce for use as an OAuth CSRF token.
Call this before :func:`oauth_build_authorization_redirect`, persist the
returned value in the user's session or a ``Secure; HttpOnly; SameSite=Lax``
cookie, then verify it with :func:`oauth_decode_state` on the callback.
"""
def oauth_generate_state_token() -> str:
"""Generate a cryptographically random CSRF token for the OAuth ``state`` parameter."""
return secrets.token_urlsafe(32)
@@ -120,7 +115,7 @@ def oauth_build_authorization_redirect(
scopes: str,
redirect_uri: str,
destination: str,
nonce: str,
state_token: str,
) -> RedirectResponse:
"""Return an OAuth 2.0 authorization ``RedirectResponse``.
@@ -131,9 +126,9 @@ def oauth_build_authorization_redirect(
redirect_uri: URI the provider should redirect back to after authorization.
destination: URL the user should be sent to after the full OAuth flow
completes (embedded in ``state``).
nonce: CSRF token generated by :func:`oauth_generate_nonce`. Must be
stored server-side (session or signed cookie) and verified via
:func:`oauth_decode_state` on the callback endpoint.
state_token: CSRF token generated by :func:`oauth_generate_state_token`.
Must be stored server-side (session or signed cookie) and verified via
:func:`oauth_decode_state` on the callback endpoint (RFC 6749 §10.12).
Returns:
A :class:`~fastapi.responses.RedirectResponse` to the provider's
@@ -145,32 +140,34 @@ def oauth_build_authorization_redirect(
"response_type": "code",
"scope": scopes,
"redirect_uri": redirect_uri,
"state": oauth_encode_state(destination, nonce),
"state": oauth_encode_state(destination, state_token),
}
)
return RedirectResponse(f"{authorization_url}?{params}")
def oauth_encode_state(url: str, nonce: str) -> str:
"""Encode a destination URL and CSRF nonce into an OAuth ``state`` parameter.
def oauth_encode_state(url: str, state_token: str) -> str:
"""Encode a destination URL and CSRF token into an OAuth ``state`` parameter.
Args:
url: Post-login destination URL.
nonce: CSRF token from :func:`oauth_generate_nonce`.
state_token: CSRF token from :func:`oauth_generate_state_token`.
"""
payload = json.dumps({"n": nonce, "d": url}, separators=(",", ":"))
payload = json.dumps({"n": state_token, "d": url}, separators=(",", ":"))
return base64.urlsafe_b64encode(payload.encode()).decode()
def oauth_decode_state(state: str | None, *, expected_nonce: str, fallback: str) -> str:
def oauth_decode_state(
state: str | None, *, expected_state_token: str, fallback: str
) -> str:
"""Decode and CSRF-verify an OAuth ``state`` parameter.
Uses a constant-time comparison for the nonce to prevent timing attacks.
Uses a constant-time comparison for the CSRF token to prevent timing attacks.
Args:
state: Raw ``state`` query parameter from the provider's callback.
expected_nonce: The nonce stored before the authorization redirect.
If the decoded nonce does not match, ``fallback`` is returned.
expected_state_token: The token stored before the authorization redirect.
If it does not match the decoded value, ``fallback`` is returned.
fallback: URL to return when ``state`` is absent, malformed, or fails
CSRF verification.
@@ -178,7 +175,7 @@ def oauth_decode_state(state: str | None, *, expected_nonce: str, fallback: str)
The destination URL embedded in ``state``, or ``fallback``.
Important:
**Single-use**: delete the stored nonce from the session immediately
**Single-use**: delete the stored token from the session immediately
after calling this function — whether it matched or not — so that a
captured callback URL cannot be replayed.
@@ -192,7 +189,7 @@ def oauth_decode_state(state: str | None, *, expected_nonce: str, fallback: str)
padded = state + "=" * (-len(state) % 4)
payload = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8"))
if not isinstance(payload, dict) or not hmac.compare_digest(
payload.get("n", "").encode(), expected_nonce.encode()
payload.get("n", "").encode(), expected_state_token.encode()
):
return fallback
return str(payload["d"])
tests/test_security.py
+31 -21
View File
@@ -18,7 +18,7 @@ from fastapi_toolsets.security import (
oauth_decode_state,
oauth_encode_state,
oauth_fetch_userinfo,
oauth_generate_nonce,
oauth_generate_state_token,
oauth_resolve_provider_urls,
)
@@ -1013,56 +1013,64 @@ def _make_async_client_mock(get_return=None, post_return=None):
class TestEncodeDecodeOAuthState:
def test_encode_returns_base64url_string(self):
result = oauth_encode_state("https://example.com/dashboard", "test-nonce")
result = oauth_encode_state("https://example.com/dashboard", "test-state-token")
assert isinstance(result, str)
assert "+" not in result
assert "/" not in result
def test_round_trip(self):
url = "https://example.com/after-login?next=/home"
nonce = "test-nonce"
state_token = "test-state-token"
assert (
oauth_decode_state(
oauth_encode_state(url, nonce), expected_nonce=nonce, fallback="/"
oauth_encode_state(url, state_token),
expected_state_token=state_token,
fallback="/",
)
== url
)
def test_decode_none_returns_fallback(self):
assert (
oauth_decode_state(None, expected_nonce="any", fallback="/home") == "/home"
oauth_decode_state(None, expected_state_token="any", fallback="/home")
== "/home"
)
def test_decode_null_string_returns_fallback(self):
assert (
oauth_decode_state("null", expected_nonce="any", fallback="/home")
oauth_decode_state("null", expected_state_token="any", fallback="/home")
== "/home"
)
def test_decode_invalid_base64_returns_fallback(self):
assert (
oauth_decode_state(
"!!!notbase64!!!", expected_nonce="any", fallback="/home"
"!!!notbase64!!!", expected_state_token="any", fallback="/home"
)
== "/home"
)
def test_decode_handles_missing_padding(self):
url = "https://example.com/x"
nonce = "test-nonce"
encoded = oauth_encode_state(url, nonce).rstrip("=")
assert oauth_decode_state(encoded, expected_nonce=nonce, fallback="/") == url
def test_decode_wrong_nonce_returns_fallback(self):
url = "https://example.com/dashboard"
encoded = oauth_encode_state(url, "correct-nonce")
state_token = "test-state-token"
encoded = oauth_encode_state(url, state_token).rstrip("=")
assert (
oauth_decode_state(encoded, expected_nonce="wrong-nonce", fallback="/")
oauth_decode_state(encoded, expected_state_token=state_token, fallback="/")
== url
)
def test_decode_wrong_state_token_returns_fallback(self):
url = "https://example.com/dashboard"
encoded = oauth_encode_state(url, "correct-token")
assert (
oauth_decode_state(
encoded, expected_state_token="wrong-token", fallback="/"
)
== "/"
)
def test_generate_nonce_is_random(self):
assert oauth_generate_nonce() != oauth_generate_nonce()
def test_generate_state_token_is_random(self):
assert oauth_generate_state_token() != oauth_generate_state_token()
class TestBuildAuthorizationRedirect:
@@ -1075,19 +1083,19 @@ class TestBuildAuthorizationRedirect:
scopes="openid email",
redirect_uri="https://app.example.com/callback",
destination="https://app.example.com/dashboard",
nonce="test-nonce",
state_token="test-state-token",
)
assert isinstance(response, RedirectResponse)
def test_redirect_location_contains_all_params(self):
nonce = "test-nonce"
state_token = "test-state-token"
response = oauth_build_authorization_redirect(
"https://auth.example.com/authorize",
client_id="my-client",
scopes="openid email",
redirect_uri="https://app.example.com/callback",
destination="https://app.example.com/dashboard",
nonce=nonce,
state_token=state_token,
)
location = response.headers["location"]
parsed = urlparse(location)
@@ -1101,7 +1109,9 @@ class TestBuildAuthorizationRedirect:
assert params["scope"] == ["openid email"]
assert params["redirect_uri"] == ["https://app.example.com/callback"]
assert (
oauth_decode_state(params["state"][0], expected_nonce=nonce, fallback="")
oauth_decode_state(
params["state"][0], expected_state_token=state_token, fallback=""
)
== "https://app.example.com/dashboard"
)