From c0d5bf89445ae52f3a71e8f4470f6115b84fa568 Mon Sep 17 00:00:00 2001 From: d3vyce Date: Tue, 30 Jun 2026 16:06:50 -0400 Subject: [PATCH 1/3] chore: remove security module --- README.md | 2 - docs/index.md | 2 - docs/module/security.md | 354 ----- docs/reference/security.md | 43 - pyproject.toml | 6 +- src/fastapi_toolsets/security/__init__.py | 26 - src/fastapi_toolsets/security/abc.py | 55 - src/fastapi_toolsets/security/oauth.py | 197 --- .../security/sources/__init__.py | 8 - .../security/sources/bearer.py | 120 -- .../security/sources/cookie.py | 148 -- .../security/sources/header.py | 67 - .../security/sources/multi.py | 71 - tests/test_security.py | 1341 ----------------- uv.lock | 11 +- 15 files changed, 3 insertions(+), 2448 deletions(-) delete mode 100644 docs/module/security.md delete mode 100644 docs/reference/security.md delete mode 100644 src/fastapi_toolsets/security/__init__.py delete mode 100644 src/fastapi_toolsets/security/abc.py delete mode 100644 src/fastapi_toolsets/security/oauth.py delete mode 100644 src/fastapi_toolsets/security/sources/__init__.py delete mode 100644 src/fastapi_toolsets/security/sources/bearer.py delete mode 100644 src/fastapi_toolsets/security/sources/cookie.py delete mode 100644 src/fastapi_toolsets/security/sources/header.py delete mode 100644 src/fastapi_toolsets/security/sources/multi.py delete mode 100644 tests/test_security.py diff --git a/README.md b/README.md index 7d61cb6b..0250824b 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,6 @@ Install only the extras you need: ```bash uv add "fastapi-toolsets[cli]" uv add "fastapi-toolsets[metrics]" -uv add "fastapi-toolsets[security]" uv add "fastapi-toolsets[pytest]" ``` @@ -57,7 +56,6 @@ uv add "fastapi-toolsets[all]" ### Optional -- **Security**: Composable authentication sources (`BearerTokenAuth`, `CookieAuth`, `APIKeyHeaderAuth`, `MultiAuth`) with HMAC-signed cookies and OAuth 2.0 / OIDC helpers - **CLI**: Django-like command-line interface with fixture management and custom commands support - **Metrics**: Prometheus metrics endpoint with provider/collector registry - **Pytest Helpers**: Async test client, database session management, `pytest-xdist` support, and table cleanup utilities diff --git a/docs/index.md b/docs/index.md index 3ef1de8a..ac923e49 100644 --- a/docs/index.md +++ b/docs/index.md @@ -31,7 +31,6 @@ Install only the extras you need: ```bash uv add "fastapi-toolsets[cli]" uv add "fastapi-toolsets[metrics]" -uv add "fastapi-toolsets[security]" uv add "fastapi-toolsets[pytest]" ``` @@ -57,7 +56,6 @@ uv add "fastapi-toolsets[all]" ### Optional -- **Security**: Composable authentication sources (`BearerTokenAuth`, `CookieAuth`, `APIKeyHeaderAuth`, `MultiAuth`) with HMAC-signed cookies and OAuth 2.0 / OIDC helpers - **CLI**: Django-like command-line interface with fixture management and custom commands support - **Metrics**: Prometheus metrics endpoint with provider/collector registry - **Pytest Helpers**: Async test client, database session management, `pytest-xdist` support, and table cleanup utilities diff --git a/docs/module/security.md b/docs/module/security.md deleted file mode 100644 index 10ebd402..00000000 --- a/docs/module/security.md +++ /dev/null @@ -1,354 +0,0 @@ -# Security - -Composable authentication helpers for FastAPI that use `Security()` for OpenAPI documentation and accept user-provided validator functions with full type flexibility. - -## Overview - -The `security` module provides four auth source classes, a `MultiAuth` factory, and a set of OAuth 2.0 / OIDC helper utilities. Each auth class wraps a FastAPI security scheme for OpenAPI and accepts a validator function called as: - -```python -await validator(credential, **kwargs) -``` - -where `kwargs` are the extra keyword arguments provided at instantiation (roles, permissions, enums, etc.). The validator returns the authenticated identity (e.g. a `User` model) which becomes the route dependency value. - -```python -from fastapi import Security -from fastapi_toolsets.security import BearerTokenAuth - -async def verify_token(token: str, *, role: str) -> User: - user = await db.get_by_token(token) - if not user or user.role != role: - raise UnauthorizedError() - return user - -bearer_admin = BearerTokenAuth(verify_token, role="admin") - -@app.get("/admin") -async def admin_route(user: User = Security(bearer_admin)): - return user -``` - -## Auth sources - -### [`BearerTokenAuth`](../reference/security.md#fastapi_toolsets.security.BearerTokenAuth) - -Reads the `Authorization: Bearer ` header. Wraps `HTTPBearer` for OpenAPI. - -```python -from fastapi_toolsets.security import BearerTokenAuth - -bearer = BearerTokenAuth(validator=verify_token) - -@app.get("/me") -async def me(user: User = Security(bearer)): - return user -``` - -#### Token prefix - -The optional `prefix` parameter restricts a `BearerTokenAuth` instance to tokens that start with a given string. The prefix is **kept** in the value passed to the validator — store and compare tokens with their prefix included. - -This lets you deploy multiple `BearerTokenAuth` instances in the same application and disambiguate them efficiently in `MultiAuth`: - -```python -user_bearer = BearerTokenAuth(verify_user, prefix="user_") # matches "Bearer user_..." -org_bearer = BearerTokenAuth(verify_org, prefix="org_") # matches "Bearer org_..." -``` - -Use [`generate_token()`](#token-generation) to create correctly-prefixed tokens. - -#### Token generation - -`BearerTokenAuth.generate_token()` produces a secure random token ready to store in your database and return to the client. If a prefix is configured it is prepended automatically: - -```python -bearer = BearerTokenAuth(verify_token, prefix="user_") - -token = bearer.generate_token() # e.g. "user_Xk3mN..." -await db.store_token(user_id, token) -return {"access_token": token, "token_type": "bearer"} -``` - -The client sends `Authorization: Bearer user_Xk3mN...` and the validator receives the full token (prefix included) to compare against the stored value. - -### [`CookieAuth`](../reference/security.md#fastapi_toolsets.security.CookieAuth) - -Reads a named cookie. Wraps `APIKeyCookie` for OpenAPI. - -Cookies are issued with the `Secure` flag set by default, meaning they are only transmitted over HTTPS. Set `secure=False` when running locally over plain HTTP: - -```python -from fastapi_toolsets.security import CookieAuth - -# Production (HTTPS) — default -cookie_auth = CookieAuth("session", validator=verify_session) - -# Local development (HTTP only) -cookie_auth = CookieAuth("session", validator=verify_session, secure=False) - -@app.get("/me") -async def me(user: User = Security(cookie_auth)): - return user -``` - -#### Signed cookies - -Pass `secret_key` to enable HMAC-SHA256 signed, tamper-proof cookies. The cookie payload includes an expiry timestamp (`ttl`, default 24 h). No database entry is required — the signature is self-contained. - -Use `set_cookie()` to issue the signed cookie on login and `delete_cookie()` to clear it on logout: - -```python -# Production -cookie_auth = CookieAuth("session", verify_session, secret_key="your-secret") - -# Local development -cookie_auth = CookieAuth("session", verify_session, secret_key="your-secret", secure=False) - -@app.post("/login") -async def login(response: Response): - cookie_auth.set_cookie(response, user_id) - return {"ok": True} - -@app.post("/logout") -async def logout(response: Response): - cookie_auth.delete_cookie(response) - return {"ok": True} - -@app.get("/me") -async def me(user: User = Security(cookie_auth)): - return user -``` - -When `secret_key` is not set, the raw cookie value is passed directly to the validator (stateful session behaviour — you manage the session store). - -### [`APIKeyHeaderAuth`](../reference/security.md#fastapi_toolsets.security.APIKeyHeaderAuth) - -Reads an API key from a named HTTP header. Wraps `APIKeyHeader` for OpenAPI. - -```python -from fastapi_toolsets.security import APIKeyHeaderAuth - -api_key_auth = APIKeyHeaderAuth("X-API-Key", validator=verify_api_key) - -@app.get("/data") -async def data(user: User = Security(api_key_auth)): - return user -``` - -The header name is configurable — use any header your API defines (e.g. `"X-API-Key"`, `"Authorization"`, `"X-Service-Token"`). - -## Typed validator kwargs - -All auth classes forward extra instantiation keyword arguments to the validator. Arguments can be any type — enums, strings, integers, etc. The validator returns the authenticated identity, which FastAPI injects directly into the route handler. - -```python -async def verify_token(token: str, *, role: Role, permission: str) -> User: - user = await decode_token(token) - if user.role != role or permission not in user.permissions: - raise UnauthorizedError() - return user - -bearer = BearerTokenAuth(verify_token, role=Role.ADMIN, permission="billing:read") -``` - -Each auth instance is self-contained — create a separate instance per distinct requirement instead of passing requirements through `Security(scopes=[...])`. - -### Using `.require()` inline - -If declaring a new top-level variable per role feels verbose, use `.require()` to create a configured clone directly in the route decorator. The original instance is not mutated: - -```python -bearer = BearerTokenAuth(verify_token) - -@app.get("/admin/stats") -async def admin_stats(user: User = Security(bearer.require(role=Role.ADMIN))): - return {"message": f"Hello admin {user.name}"} - -@app.get("/profile") -async def profile(user: User = Security(bearer.require(role=Role.USER))): - return {"id": user.id, "name": user.name} -``` - -`.require()` kwargs are merged over existing ones — new values win on conflict. -The `prefix` (for `BearerTokenAuth`), cookie name and `secret_key` (for -`CookieAuth`), and header name (for `APIKeyHeaderAuth`) are always preserved. - -## 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. - -If a credential is extracted but the validator raises, the exception propagates immediately — the remaining sources are **not** tried. This prevents silent fallthrough on invalid credentials. - -```python -from fastapi_toolsets.security import MultiAuth - -multi = MultiAuth(user_bearer, org_bearer, cookie_auth) - -@app.get("/data") -async def data_route(user = Security(multi)): - return user -``` - -### Using `.require()` on MultiAuth - -`MultiAuth` also supports `.require()`, which propagates the kwargs to every source that implements it. Sources that do not (e.g. custom `AuthSource` subclasses) are passed through unchanged: - -```python -multi = MultiAuth(bearer, cookie) - -@app.get("/admin") -async def admin(user: User = Security(multi.require(role=Role.ADMIN))): - return user -``` - -This is equivalent to calling `.require()` on each source individually: - -```python -# These two are identical -multi.require(role=Role.ADMIN) - -MultiAuth( - bearer.require(role=Role.ADMIN), - cookie.require(role=Role.ADMIN), -) -``` - -### Prefix-based dispatch - -Because `extract()` is pure string matching (no I/O), prefix-based source selection is essentially free. Only the matching source's validator (which may involve DB or network I/O) is ever called: - -```python -user_bearer = BearerTokenAuth(verify_user, prefix="user_") -org_bearer = BearerTokenAuth(verify_org, prefix="org_") - -multi = MultiAuth(user_bearer, org_bearer) - -# "Bearer user_alice" → only verify_user runs, receives "user_alice" -# "Bearer org_acme" → only verify_org runs, receives "org_acme" -``` - -Tokens are stored and compared **with their prefix** — use `generate_token()` on each source to issue correctly-prefixed tokens: - -```python -user_token = user_bearer.generate_token() # "user_..." -org_token = org_bearer.generate_token() # "org_..." -``` - -## Custom auth sources - -Subclass [`AuthSource`](../reference/security.md#fastapi_toolsets.security.AuthSource) to implement any credential extraction strategy. You only need to implement `extract()` and `authenticate()`: - -```python -from fastapi_toolsets.security import AuthSource -from fastapi_toolsets.exceptions import UnauthorizedError - -class MTLSAuth(AuthSource): - async def extract(self, request) -> str | None: - return request.headers.get("X-Client-Cert-DN") or None - - async def authenticate(self, credential: str): - dn = parse_dn(credential) - if dn.get("O") != "MyOrg": - raise UnauthorizedError() - return {"dn": credential} -``` - -Custom sources work transparently inside `MultiAuth`. - -## OAuth 2.0 / OIDC helpers - -The module provides standalone async utilities for building OAuth 2.0 / OIDC login flows. They handle provider discovery, authorization redirects, token exchange, and state encoding — leaving JWT validation and session management to your application. - -### Provider discovery - -[`oauth_resolve_provider_urls()`](../reference/security.md#fastapi_toolsets.security.oauth_resolve_provider_urls) fetches the OIDC discovery document and returns the endpoint URLs. Results are cached in-process to avoid repeated network calls: - -```python -from fastapi_toolsets.security import oauth_resolve_provider_urls - -auth_url, token_url, userinfo_url = await oauth_resolve_provider_urls( - "https://accounts.google.com/.well-known/openid-configuration" -) -``` - -Returns a `(authorization_url, token_url, userinfo_url)` tuple. `userinfo_url` is `None` when the provider does not advertise one. - -### 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 `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_state_token - -@app.get("/auth/google/login") -async def google_login(request: Request): - auth_url, _, _ = await oauth_resolve_provider_urls(GOOGLE_DISCOVERY_URL) - 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", - state_token=state_token, - ) -``` - -### Token exchange and userinfo - -[`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 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 -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 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_state_token=state_token, fallback="/") - if not destination.startswith("/"): # reject absolute URLs to prevent open-redirect - destination = "/" - - _, token_url, userinfo_url = await oauth_resolve_provider_urls(GOOGLE_DISCOVERY_URL) - userinfo = await oauth_fetch_userinfo( - token_url=token_url, - userinfo_url=userinfo_url, - code=code, - client_id=GOOGLE_CLIENT_ID, - client_secret=GOOGLE_CLIENT_SECRET, - redirect_uri="https://myapp.com/auth/google/callback", - required_scopes="openid email profile", - ) - user = await db.upsert_user(email=userinfo["email"]) - response = RedirectResponse(destination) - session_cookie.set_cookie(response, str(user.id)) - return response -``` - -Pass `required_scopes` to guard against providers silently granting fewer scopes than requested — `oauth_fetch_userinfo` raises `ValueError` if any are missing. - -### 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 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 - -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="/") # "/" -``` - ---- - -[:material-api: API Reference](../reference/security.md) diff --git a/docs/reference/security.md b/docs/reference/security.md deleted file mode 100644 index 73b37f36..00000000 --- a/docs/reference/security.md +++ /dev/null @@ -1,43 +0,0 @@ -# `security` - -Here's the reference for the authentication helpers provided by the `security` module. - -You can import them directly from `fastapi_toolsets.security`: - -```python -from fastapi_toolsets.security import ( - AuthSource, - BearerTokenAuth, - CookieAuth, - APIKeyHeaderAuth, - MultiAuth, - oauth_build_authorization_redirect, - oauth_decode_state, - oauth_encode_state, - oauth_fetch_userinfo, - oauth_generate_state_token, - oauth_resolve_provider_urls, -) -``` - -## ::: fastapi_toolsets.security.AuthSource - -## ::: fastapi_toolsets.security.BearerTokenAuth - -## ::: fastapi_toolsets.security.CookieAuth - -## ::: fastapi_toolsets.security.APIKeyHeaderAuth - -## ::: fastapi_toolsets.security.MultiAuth - -## ::: fastapi_toolsets.security.oauth_resolve_provider_urls - -## ::: fastapi_toolsets.security.oauth_fetch_userinfo - -## ::: fastapi_toolsets.security.oauth_generate_state_token - -## ::: fastapi_toolsets.security.oauth_build_authorization_redirect - -## ::: fastapi_toolsets.security.oauth_encode_state - -## ::: fastapi_toolsets.security.oauth_decode_state diff --git a/pyproject.toml b/pyproject.toml index 03ebed90..3806f625 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,17 +50,13 @@ cli = [ metrics = [ "prometheus_client>=0.20.0", ] -security = [ - "async-lru>=1.0", - "httpx>=0.25.0", -] pytest = [ "httpx>=0.25.0", "pytest-xdist>=3.0.0", "pytest>=8.0.0", ] all = [ - "fastapi-toolsets[cli,metrics,pytest,security]", + "fastapi-toolsets[cli,metrics,pytest]", ] [project.scripts] diff --git a/src/fastapi_toolsets/security/__init__.py b/src/fastapi_toolsets/security/__init__.py deleted file mode 100644 index 138b94e7..00000000 --- a/src/fastapi_toolsets/security/__init__.py +++ /dev/null @@ -1,26 +0,0 @@ -"""Authentication helpers for FastAPI using Security().""" - -from .abc import AuthSource -from .oauth import ( - oauth_build_authorization_redirect, - oauth_decode_state, - oauth_encode_state, - oauth_fetch_userinfo, - oauth_generate_state_token, - oauth_resolve_provider_urls, -) -from .sources import APIKeyHeaderAuth, BearerTokenAuth, CookieAuth, MultiAuth - -__all__ = [ - "APIKeyHeaderAuth", - "AuthSource", - "BearerTokenAuth", - "CookieAuth", - "MultiAuth", - "oauth_build_authorization_redirect", - "oauth_decode_state", - "oauth_encode_state", - "oauth_fetch_userinfo", - "oauth_generate_state_token", - "oauth_resolve_provider_urls", -] diff --git a/src/fastapi_toolsets/security/abc.py b/src/fastapi_toolsets/security/abc.py deleted file mode 100644 index 9eb8c93a..00000000 --- a/src/fastapi_toolsets/security/abc.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Abstract base class for authentication sources.""" - -import functools -import inspect -from abc import ABC, abstractmethod -from typing import Any, Callable - -from fastapi import Request -from fastapi.security import SecurityScopes - -from fastapi_toolsets.exceptions import UnauthorizedError - - -def _ensure_async(fn: Callable[..., Any]) -> Callable[..., Any]: - """Wrap *fn* so it can always be awaited, caching the coroutine check at init time.""" - if inspect.iscoroutinefunction(fn): - return fn - - @functools.wraps(fn) - async def wrapper(*args: Any, **kwargs: Any) -> Any: - return fn(*args, **kwargs) - - return wrapper - - -class AuthSource(ABC): - """Abstract base class for authentication sources.""" - - def __init__(self) -> None: - """Set up the default FastAPI dependency signature.""" - source = self - - async def _call( - request: Request, - security_scopes: SecurityScopes, # noqa: ARG001 - ) -> Any: - credential = await source.extract(request) - if credential is None: - raise UnauthorizedError() - return await source.authenticate(credential) - - self._call_fn: Callable[..., Any] = _call - self.__signature__ = inspect.signature(_call) - - @abstractmethod - async def extract(self, request: Request) -> str | None: - """Extract the raw credential from the request without validating.""" - - @abstractmethod - async def authenticate(self, credential: str) -> Any: - """Validate a credential and return the authenticated identity.""" - - async def __call__(self, **kwargs: Any) -> Any: - """FastAPI dependency dispatch.""" - return await self._call_fn(**kwargs) diff --git a/src/fastapi_toolsets/security/oauth.py b/src/fastapi_toolsets/security/oauth.py deleted file mode 100644 index 4625ae12..00000000 --- a/src/fastapi_toolsets/security/oauth.py +++ /dev/null @@ -1,197 +0,0 @@ -"""OAuth 2.0 / OIDC helper utilities.""" - -import base64 -import binascii -import hmac -import json -import secrets -from typing import Any -from urllib.parse import urlencode - -import httpx -from async_lru import alru_cache -from fastapi.responses import RedirectResponse - - -@alru_cache(maxsize=32) -async def oauth_resolve_provider_urls( - discovery_url: str, -) -> tuple[str, str, str | None]: - """Fetch the OIDC discovery document and return endpoint URLs. - - Args: - discovery_url: URL of the provider's ``/.well-known/openid-configuration``. - - Returns: - A ``(authorization_url, token_url, userinfo_url)`` tuple. - *userinfo_url* is ``None`` when the provider does not advertise one. - """ - async with httpx.AsyncClient() as client: - resp = await client.get(discovery_url) - resp.raise_for_status() - cfg = resp.json() - return ( - cfg["authorization_endpoint"], - cfg["token_endpoint"], - cfg.get("userinfo_endpoint"), - ) - - -async def oauth_fetch_userinfo( - *, - token_url: str, - userinfo_url: str, - code: str, - client_id: str, - client_secret: str, - redirect_uri: str, - required_scopes: str | None = None, -) -> dict[str, Any]: - """Exchange an authorization code for tokens and return the userinfo payload. - - Args: - token_url: Provider's token endpoint. - userinfo_url: Provider's userinfo endpoint. - code: Authorization code received from the provider's callback. - client_id: OAuth application client ID. - client_secret: OAuth application client secret. - redirect_uri: Redirect URI that was used in the authorization request. - required_scopes: Space-separated scopes that must be present in the token - response ``scope`` field (RFC 6749 §3.3). Raises ``ValueError`` if - the provider granted fewer scopes than requested. - - Returns: - The JSON payload returned by the userinfo endpoint as a plain ``dict``. - - Raises: - ValueError: If the provider granted a different token type than ``bearer`` - or did not grant all ``required_scopes``. - """ - async with httpx.AsyncClient() as client: - token_resp = await client.post( - token_url, - data={ - "grant_type": "authorization_code", - "code": code, - "client_id": client_id, - "client_secret": client_secret, - "redirect_uri": redirect_uri, - }, - headers={"Accept": "application/json"}, - ) - token_resp.raise_for_status() - token_data = token_resp.json() - - if token_data.get("token_type", "bearer").lower() != "bearer": - raise ValueError( - f"unsupported token_type: {token_data.get('token_type')!r}" - ) - - if required_scopes is not None: - granted = set(token_data.get("scope", "").split()) - missing = set(required_scopes.split()) - granted - if missing: - raise ValueError(f"provider did not grant required scopes: {missing}") - - access_token = token_data["access_token"] - - userinfo_resp = await client.get( - userinfo_url, - headers={"Authorization": f"Bearer {access_token}"}, - ) - userinfo_resp.raise_for_status() - return userinfo_resp.json() - - -def oauth_generate_state_token() -> str: - """Generate a cryptographically random CSRF token for the OAuth ``state`` parameter.""" - return secrets.token_urlsafe(32) - - -def oauth_build_authorization_redirect( - authorization_url: str, - *, - client_id: str, - scopes: str, - redirect_uri: str, - destination: str, - state_token: str, -) -> RedirectResponse: - """Return an OAuth 2.0 authorization ``RedirectResponse``. - - Args: - authorization_url: Provider's authorization endpoint. - client_id: OAuth application client ID. - scopes: Space-separated list of requested scopes. - 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``). - 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 - authorization page. - """ - params = urlencode( - { - "client_id": client_id, - "response_type": "code", - "scope": scopes, - "redirect_uri": redirect_uri, - "state": oauth_encode_state(destination, state_token), - } - ) - return RedirectResponse(f"{authorization_url}?{params}") - - -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. - state_token: CSRF token from :func:`oauth_generate_state_token`. - """ - payload = json.dumps({"n": state_token, "d": url}, separators=(",", ":")) - return base64.urlsafe_b64encode(payload.encode()).decode() - - -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 CSRF token to prevent timing attacks. - - Args: - state: Raw ``state`` query parameter from the provider's callback. - 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. - - Returns: - The destination URL embedded in ``state``, or ``fallback``. - - Important: - **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. - - **Open-redirect**: validate the returned URL against a known-good - origin or relative-path allowlist before issuing the final redirect. - Do not forward arbitrary URLs to ``RedirectResponse``. - """ - if not state or state == "null": # "null" guards against JS JSON.stringify(null) - return fallback - try: - 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_state_token.encode() - ): - return fallback - return str(payload["d"]) - except (UnicodeDecodeError, ValueError, binascii.Error, KeyError): - return fallback diff --git a/src/fastapi_toolsets/security/sources/__init__.py b/src/fastapi_toolsets/security/sources/__init__.py deleted file mode 100644 index 8f90c545..00000000 --- a/src/fastapi_toolsets/security/sources/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -"""Built-in authentication source implementations.""" - -from .header import APIKeyHeaderAuth -from .bearer import BearerTokenAuth -from .cookie import CookieAuth -from .multi import MultiAuth - -__all__ = ["APIKeyHeaderAuth", "BearerTokenAuth", "CookieAuth", "MultiAuth"] diff --git a/src/fastapi_toolsets/security/sources/bearer.py b/src/fastapi_toolsets/security/sources/bearer.py deleted file mode 100644 index b33f4325..00000000 --- a/src/fastapi_toolsets/security/sources/bearer.py +++ /dev/null @@ -1,120 +0,0 @@ -"""Bearer token authentication source.""" - -import inspect -import secrets -from typing import Annotated, Any, Callable - -from fastapi import Depends, Request -from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer, SecurityScopes - -from fastapi_toolsets.exceptions import UnauthorizedError - -from ..abc import AuthSource, _ensure_async - - -class BearerTokenAuth(AuthSource): - """Bearer token authentication source. - - Wraps :class:`fastapi.security.HTTPBearer` for OpenAPI documentation. - The validator is called as ``await validator(credential, **kwargs)`` - where ``kwargs`` are the extra keyword arguments provided at instantiation. - - Args: - validator: Sync or async callable that receives the credential and any - extra keyword arguments, and returns the authenticated identity - (e.g. a ``User`` model). Should raise - :class:`~fastapi_toolsets.exceptions.UnauthorizedError` on failure. - prefix: Optional token prefix (e.g. ``"user_"``). If set, only tokens - whose value starts with this prefix are matched. The prefix is - **kept** in the value passed to the validator — store and compare - tokens with their prefix included. Use :meth:`generate_token` to - create correctly-prefixed tokens. This enables multiple - ``BearerTokenAuth`` instances in the same app (e.g. ``"user_"`` - for user tokens, ``"org_"`` for org tokens). - **kwargs: Extra keyword arguments forwarded to the validator on every - call (e.g. ``role=Role.ADMIN``). - """ - - def __init__( - self, - validator: Callable[..., Any], - *, - prefix: str | None = None, - **kwargs: Any, - ) -> None: - self._validator = _ensure_async(validator) - self._prefix = prefix - self._kwargs = kwargs - self._scheme = HTTPBearer(auto_error=False) - - async def _call( - security_scopes: SecurityScopes, # noqa: ARG001 - credentials: Annotated[ - HTTPAuthorizationCredentials | None, Depends(self._scheme) - ] = None, - ) -> Any: - if credentials is None: - raise UnauthorizedError() - return await self._validate(credentials.credentials) - - self._call_fn = _call - self.__signature__ = inspect.signature(_call) - - async def _validate(self, token: str) -> Any: - """Check prefix and call the validator.""" - if self._prefix is not None and not token.startswith(self._prefix): - raise UnauthorizedError() - return await self._validator(token, **self._kwargs) - - async def extract(self, request: Request) -> str | None: - """Extract the raw credential from the request without validating. - - Returns ``None`` if no ``Authorization: Bearer`` header is present, - the token is empty, or the token does not match the configured prefix. - The prefix is included in the returned value. - """ - auth = request.headers.get("Authorization", "") - if not auth.startswith("Bearer "): - return None - token = auth[7:] - if not token: - return None - if self._prefix is not None and not token.startswith(self._prefix): - return None - return token - - async def authenticate(self, credential: str) -> Any: - """Validate a credential and return the identity. - - Calls ``await validator(credential, **kwargs)`` where ``kwargs`` are - the extra keyword arguments provided at instantiation. - """ - return await self._validate(credential) - - def require(self, **kwargs: Any) -> "BearerTokenAuth": - """Return a new instance with additional (or overriding) validator kwargs.""" - return BearerTokenAuth( - self._validator, - prefix=self._prefix, - **{**self._kwargs, **kwargs}, - ) - - def generate_token(self, nbytes: int = 32) -> str: - """Generate a secure random token for this auth source. - - Returns a URL-safe random token. If a prefix is configured it is - prepended — the returned value is what you store in your database - and return to the client as-is. - - Args: - nbytes: Number of random bytes before base64 encoding. The - resulting string is ``ceil(nbytes * 4 / 3)`` characters - (43 chars for the default 32 bytes). Defaults to 32. - - Returns: - A ready-to-use token string (e.g. ``"user_Xk3..."``). - """ - token = secrets.token_urlsafe(nbytes) - if self._prefix is not None: - return f"{self._prefix}{token}" - return token diff --git a/src/fastapi_toolsets/security/sources/cookie.py b/src/fastapi_toolsets/security/sources/cookie.py deleted file mode 100644 index 9fd69046..00000000 --- a/src/fastapi_toolsets/security/sources/cookie.py +++ /dev/null @@ -1,148 +0,0 @@ -"""Cookie-based authentication source.""" - -import base64 -import hashlib -import hmac -import inspect -import json -import time -from typing import Annotated, Any, Callable - -from fastapi import Depends, Request, Response -from fastapi.security import APIKeyCookie, SecurityScopes - -from fastapi_toolsets.exceptions import UnauthorizedError - -from ..abc import AuthSource, _ensure_async - - -class CookieAuth(AuthSource): - """Cookie-based authentication source. - - Wraps :class:`fastapi.security.APIKeyCookie` for OpenAPI documentation. - Optionally signs the cookie with HMAC-SHA256 to provide stateless, tamper- - proof sessions without any database entry. - - Args: - name: Cookie name. - validator: Sync or async callable that receives the cookie value - (plain, after signature verification when ``secret_key`` is set) - and any extra keyword arguments, and returns the authenticated - identity. - secret_key: When provided, the cookie is HMAC-SHA256 signed. - :meth:`set_cookie` embeds an expiry and signs the payload; - :meth:`extract` verifies the signature and expiry before handing - the plain value to the validator. When ``None`` (default), the raw - cookie value is passed to the validator as-is. - ttl: Cookie lifetime in seconds (default 24 h). Only used when - ``secret_key`` is set. - secure: Set the ``Secure`` flag on the cookie so it is only transmitted - over HTTPS (default ``True``). Set to ``False`` only in local - development environments where HTTPS is unavailable. - **kwargs: Extra keyword arguments forwarded to the validator on every - call (e.g. ``role=Role.ADMIN``). - """ - - def __init__( - self, - name: str, - validator: Callable[..., Any], - *, - secret_key: str | None = None, - ttl: int = 86400, - secure: bool = True, - **kwargs: Any, - ) -> None: - self._name = name - self._validator = _ensure_async(validator) - self._secret_key = secret_key - self._ttl = ttl - self._secure = secure - self._kwargs = kwargs - self._scheme = APIKeyCookie(name=name, auto_error=False) - - async def _call( - security_scopes: SecurityScopes, # noqa: ARG001 - value: Annotated[str | None, Depends(self._scheme)] = None, - ) -> Any: - if value is None: - raise UnauthorizedError() - plain = self._verify(value) - return await self._validator(plain, **self._kwargs) - - self._call_fn = _call - self.__signature__ = inspect.signature(_call) - - def _hmac(self, data: str) -> str: - if self._secret_key is None: - raise RuntimeError("_hmac called without secret_key configured") - return hmac.new( - self._secret_key.encode(), data.encode(), hashlib.sha256 - ).hexdigest() - - def _sign(self, value: str) -> str: - data = base64.urlsafe_b64encode( - json.dumps({"v": value, "exp": int(time.time()) + self._ttl}).encode() - ).decode() - return f"{data}.{self._hmac(data)}" - - def _verify(self, cookie_value: str) -> str: - """Return the plain value, verifying HMAC + expiry when signed.""" - if not self._secret_key: - return cookie_value - - try: - data, sig = cookie_value.rsplit(".", 1) - except ValueError: - raise UnauthorizedError() - - if not hmac.compare_digest(self._hmac(data), sig): - raise UnauthorizedError() - - try: - payload = json.loads(base64.urlsafe_b64decode(data)) - value: str = payload["v"] - exp: int = payload["exp"] - except Exception: - raise UnauthorizedError() - - if exp < int(time.time()): - raise UnauthorizedError() - - return value - - async def extract(self, request: Request) -> str | None: - return request.cookies.get(self._name) - - async def authenticate(self, credential: str) -> Any: - plain = self._verify(credential) - return await self._validator(plain, **self._kwargs) - - def require(self, **kwargs: Any) -> "CookieAuth": - """Return a new instance with additional (or overriding) validator kwargs.""" - return CookieAuth( - self._name, - self._validator, - secret_key=self._secret_key, - ttl=self._ttl, - secure=self._secure, - **{**self._kwargs, **kwargs}, - ) - - def set_cookie(self, response: Response, value: str) -> None: - """Attach the cookie to *response*, signing it when ``secret_key`` is set.""" - cookie_value = self._sign(value) if self._secret_key else value - response.set_cookie( - self._name, - cookie_value, - httponly=True, - samesite="lax", - secure=self._secure, - max_age=self._ttl, - ) - - def delete_cookie(self, response: Response) -> None: - """Clear the session cookie (logout).""" - response.delete_cookie( - self._name, httponly=True, samesite="lax", secure=self._secure - ) diff --git a/src/fastapi_toolsets/security/sources/header.py b/src/fastapi_toolsets/security/sources/header.py deleted file mode 100644 index ec4834f2..00000000 --- a/src/fastapi_toolsets/security/sources/header.py +++ /dev/null @@ -1,67 +0,0 @@ -"""API key header authentication source.""" - -import inspect -from typing import Annotated, Any, Callable - -from fastapi import Depends, Request -from fastapi.security import APIKeyHeader, SecurityScopes - -from fastapi_toolsets.exceptions import UnauthorizedError - -from ..abc import AuthSource, _ensure_async - - -class APIKeyHeaderAuth(AuthSource): - """API key header authentication source. - - Wraps :class:`fastapi.security.APIKeyHeader` for OpenAPI documentation. - The validator is called as ``await validator(api_key, **kwargs)`` - where ``kwargs`` are the extra keyword arguments provided at instantiation. - - Args: - name: HTTP header name that carries the API key (e.g. ``"X-API-Key"``). - validator: Sync or async callable that receives the API key and any - extra keyword arguments, and returns the authenticated identity. - Should raise :class:`~fastapi_toolsets.exceptions.UnauthorizedError` - on failure. - **kwargs: Extra keyword arguments forwarded to the validator on every - call (e.g. ``role=Role.ADMIN``). - """ - - def __init__( - self, - name: str, - validator: Callable[..., Any], - **kwargs: Any, - ) -> None: - self._name = name - self._validator = _ensure_async(validator) - self._kwargs = kwargs - self._scheme = APIKeyHeader(name=name, auto_error=False) - - async def _call( - security_scopes: SecurityScopes, # noqa: ARG001 - api_key: Annotated[str | None, Depends(self._scheme)] = None, - ) -> Any: - if api_key is None: - raise UnauthorizedError() - return await self._validator(api_key, **self._kwargs) - - self._call_fn = _call - self.__signature__ = inspect.signature(_call) - - async def extract(self, request: Request) -> str | None: - """Extract the API key from the configured header.""" - return request.headers.get(self._name) or None - - async def authenticate(self, credential: str) -> Any: - """Validate a credential and return the identity.""" - return await self._validator(credential, **self._kwargs) - - def require(self, **kwargs: Any) -> "APIKeyHeaderAuth": - """Return a new instance with additional (or overriding) validator kwargs.""" - return APIKeyHeaderAuth( - self._name, - self._validator, - **{**self._kwargs, **kwargs}, - ) diff --git a/src/fastapi_toolsets/security/sources/multi.py b/src/fastapi_toolsets/security/sources/multi.py deleted file mode 100644 index 5180b115..00000000 --- a/src/fastapi_toolsets/security/sources/multi.py +++ /dev/null @@ -1,71 +0,0 @@ -"""MultiAuth: combine multiple authentication sources into a single callable.""" - -import inspect -from typing import Any, cast - -from fastapi import Request -from fastapi.security import SecurityScopes - -from fastapi_toolsets.exceptions import UnauthorizedError - -from ..abc import AuthSource - - -class MultiAuth: - """Combine multiple authentication sources into a single callable. - - Args: - *sources: Auth source instances to try in order. - """ - - def __init__(self, *sources: AuthSource) -> None: - self._sources = sources - - async def _call( - request: Request, - security_scopes: SecurityScopes, # noqa: ARG001 - **kwargs: Any, # noqa: ARG001 — absorbs scheme values injected by FastAPI - ) -> Any: - for source in self._sources: - credential = await source.extract(request) - if credential is not None: - return await source.authenticate(credential) - raise UnauthorizedError() - - self._call_fn = _call - - # Build a merged signature that includes the security-scheme Depends() - # parameters from every source so FastAPI registers them in OpenAPI docs. - seen: set[str] = {"request", "security_scopes"} - merged: list[inspect.Parameter] = [ - inspect.Parameter( - "request", - inspect.Parameter.POSITIONAL_OR_KEYWORD, - annotation=Request, - ), - inspect.Parameter( - "security_scopes", - inspect.Parameter.POSITIONAL_OR_KEYWORD, - annotation=SecurityScopes, - ), - ] - for i, source in enumerate(sources): - for name, param in inspect.signature(source).parameters.items(): - if name in seen: - continue - merged.append(param.replace(name=f"_s{i}_{name}")) - seen.add(name) - self.__signature__ = inspect.Signature(merged, return_annotation=Any) - - async def __call__(self, **kwargs: Any) -> Any: - return await self._call_fn(**kwargs) - - def require(self, **kwargs: Any) -> "MultiAuth": - """Return a new :class:`MultiAuth` with kwargs forwarded to each source.""" - new_sources = tuple( - cast(Any, source).require(**kwargs) - if hasattr(source, "require") - else source - for source in self._sources - ) - return MultiAuth(*new_sources) diff --git a/tests/test_security.py b/tests/test_security.py deleted file mode 100644 index 6887cf47..00000000 --- a/tests/test_security.py +++ /dev/null @@ -1,1341 +0,0 @@ -"""Tests for fastapi_toolsets.security.""" - -from unittest.mock import AsyncMock, MagicMock, patch -from urllib.parse import parse_qs, urlparse - -import pytest -from fastapi import FastAPI, Security -from fastapi.testclient import TestClient - -from fastapi_toolsets.exceptions import UnauthorizedError, init_exceptions_handlers -from fastapi_toolsets.security import ( - APIKeyHeaderAuth, - AuthSource, - BearerTokenAuth, - CookieAuth, - MultiAuth, - oauth_build_authorization_redirect, - oauth_decode_state, - oauth_encode_state, - oauth_fetch_userinfo, - oauth_generate_state_token, - oauth_resolve_provider_urls, -) - - -def _app(*routes_setup_fns): - """Build a minimal FastAPI test app with exception handlers.""" - app = FastAPI() - init_exceptions_handlers(app) - for fn in routes_setup_fns: - fn(app) - return app - - -VALID_TOKEN = "secret" -VALID_COOKIE = "session123" - - -async def simple_validator(credential: str) -> dict: - if credential != VALID_TOKEN: - raise UnauthorizedError() - return {"user": "alice"} - - -async def role_validator(credential: str, *, role: str) -> dict: - if credential != VALID_TOKEN: - raise UnauthorizedError() - return {"user": "alice", "role": role} - - -async def cookie_validator(value: str) -> dict: - if value != VALID_COOKIE: - raise UnauthorizedError() - return {"session": value} - - -class TestBearerTokenAuth: - def test_valid_token_returns_identity(self): - bearer = BearerTokenAuth(simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - def test_missing_header_returns_401(self): - bearer = BearerTokenAuth(simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me") - assert response.status_code == 401 - - def test_invalid_token_returns_401(self): - bearer = BearerTokenAuth(simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": "Bearer wrong"}) - assert response.status_code == 401 - - def test_kwargs_forwarded_to_validator(self): - bearer = BearerTokenAuth(role_validator, role="admin") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - def test_prefix_matching_passes_full_token(self): - """Token with matching prefix: full token (with prefix) is passed to validator.""" - received: list[str] = [] - - async def capturing_validator(credential: str) -> dict: - received.append(credential) - return {"user": "alice"} - - bearer = BearerTokenAuth(capturing_validator, prefix="user_") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": "Bearer user_abc123"}) - assert response.status_code == 200 - # Prefix is kept — validator receives the full token as stored in DB - assert received == ["user_abc123"] - - def test_prefix_mismatch_returns_401(self): - bearer = BearerTokenAuth(simple_validator, prefix="user_") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": "Bearer org_abc123"}) - assert response.status_code == 401 - - @pytest.mark.anyio - async def test_extract_no_header(self): - from starlette.requests import Request - - bearer = BearerTokenAuth(simple_validator) - scope = {"type": "http", "method": "GET", "path": "/", "headers": []} - request = Request(scope) - assert await bearer.extract(request) is None - - @pytest.mark.anyio - async def test_extract_empty_token(self): - from starlette.requests import Request - - bearer = BearerTokenAuth(simple_validator) - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"authorization", b"Bearer ")], - } - request = Request(scope) - assert await bearer.extract(request) is None - - @pytest.mark.anyio - async def test_extract_no_prefix(self): - from starlette.requests import Request - - bearer = BearerTokenAuth(simple_validator) - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"authorization", b"Bearer mytoken")], - } - request = Request(scope) - assert await bearer.extract(request) == "mytoken" - - @pytest.mark.anyio - async def test_extract_prefix_match(self): - from starlette.requests import Request - - bearer = BearerTokenAuth(simple_validator, prefix="user_") - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"authorization", b"Bearer user_abc")], - } - request = Request(scope) - assert await bearer.extract(request) == "user_abc" - - @pytest.mark.anyio - async def test_extract_prefix_no_match(self): - from starlette.requests import Request - - bearer = BearerTokenAuth(simple_validator, prefix="user_") - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"authorization", b"Bearer org_abc")], - } - request = Request(scope) - assert await bearer.extract(request) is None - - def test_generate_token_no_prefix(self): - bearer = BearerTokenAuth(simple_validator) - token = bearer.generate_token() - assert isinstance(token, str) - assert len(token) > 0 - - def test_generate_token_with_prefix(self): - bearer = BearerTokenAuth(simple_validator, prefix="user_") - token = bearer.generate_token() - assert token.startswith("user_") - - def test_generate_token_uniqueness(self): - bearer = BearerTokenAuth(simple_validator) - assert bearer.generate_token() != bearer.generate_token() - - def test_generate_token_is_valid_credential(self): - """A generated token (with prefix) is accepted by the same auth source.""" - stored: list[str] = [] - - async def storing_validator(credential: str) -> dict: - stored.append(credential) - return {"token": credential} - - bearer = BearerTokenAuth(storing_validator, prefix="user_") - token = bearer.generate_token() - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {token}"}) - assert response.status_code == 200 - assert stored == [token] - - -class TestCookieAuth: - def test_valid_cookie_returns_identity(self): - cookie_auth = CookieAuth("session", cookie_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(cookie_auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", cookies={"session": VALID_COOKIE}) - assert response.status_code == 200 - assert response.json() == {"session": VALID_COOKIE} - - def test_missing_cookie_returns_401(self): - cookie_auth = CookieAuth("session", cookie_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(cookie_auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me") - assert response.status_code == 401 - - def test_invalid_cookie_returns_401(self): - cookie_auth = CookieAuth("session", cookie_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(cookie_auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", cookies={"session": "wrong"}) - assert response.status_code == 401 - - def test_kwargs_forwarded_to_validator(self): - async def session_validator(value: str, *, scope: str) -> dict: - if value != VALID_COOKIE: - raise UnauthorizedError() - return {"session": value, "scope": scope} - - cookie_auth = CookieAuth("session", session_validator, scope="read") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(cookie_auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", cookies={"session": VALID_COOKIE}) - assert response.status_code == 200 - assert response.json() == {"session": VALID_COOKIE, "scope": "read"} - - @pytest.mark.anyio - async def test_extract_no_cookie(self): - from starlette.requests import Request - - auth = CookieAuth("session", cookie_validator) - scope = {"type": "http", "method": "GET", "path": "/", "headers": []} - request = Request(scope) - assert await auth.extract(request) is None - - @pytest.mark.anyio - async def test_extract_cookie_present(self): - from starlette.requests import Request - - auth = CookieAuth("session", cookie_validator) - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"cookie", b"session=abc")], - } - request = Request(scope) - assert await auth.extract(request) == "abc" - - -class TestAPIKeyHeaderAuth: - def test_valid_key_returns_identity(self): - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"X-API-Key": VALID_TOKEN}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - def test_missing_header_returns_401(self): - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me") - assert response.status_code == 401 - - def test_invalid_key_returns_401(self): - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"X-API-Key": "wrong"}) - assert response.status_code == 401 - - def test_kwargs_forwarded_to_validator(self): - auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="admin") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"X-API-Key": VALID_TOKEN}) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - def test_require_forwards_kwargs(self): - auth = APIKeyHeaderAuth("X-API-Key", role_validator) - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(auth.require(role="admin"))): - return user - - client = TestClient(_app(setup)) - response = client.get("/admin", headers={"X-API-Key": VALID_TOKEN}) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - def test_require_preserves_name(self): - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - derived = auth.require(role="admin") - assert derived._name == "X-API-Key" - - def test_require_does_not_mutate_original(self): - auth = APIKeyHeaderAuth("X-API-Key", role_validator, role="user") - auth.require(role="admin") - assert auth._kwargs == {"role": "user"} - - def test_in_multi_auth(self): - """APIKeyHeaderAuth.authenticate() is exercised inside MultiAuth.""" - bearer = BearerTokenAuth(simple_validator) - api_key = APIKeyHeaderAuth("X-API-Key", simple_validator) - multi = MultiAuth(bearer, api_key) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - # No bearer → falls through to API key header - response = client.get("/me", headers={"X-API-Key": VALID_TOKEN}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - def test_is_auth_source(self): - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - assert isinstance(auth, AuthSource) - - @pytest.mark.anyio - async def test_extract_no_header(self): - from starlette.requests import Request - - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - scope = {"type": "http", "method": "GET", "path": "/", "headers": []} - request = Request(scope) - assert await auth.extract(request) is None - - @pytest.mark.anyio - async def test_extract_empty_header(self): - from starlette.requests import Request - - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"x-api-key", b"")], - } - request = Request(scope) - assert await auth.extract(request) is None - - @pytest.mark.anyio - async def test_extract_key_present(self): - from starlette.requests import Request - - auth = APIKeyHeaderAuth("X-API-Key", simple_validator) - scope = { - "type": "http", - "method": "GET", - "path": "/", - "headers": [(b"x-api-key", b"mykey")], - } - request = Request(scope) - assert await auth.extract(request) == "mykey" - - -class TestMultiAuth: - def test_first_source_matches(self): - bearer = BearerTokenAuth(simple_validator) - cookie = CookieAuth("session", cookie_validator) - multi = MultiAuth(bearer, cookie) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - def test_second_source_matches_when_first_absent(self): - bearer = BearerTokenAuth(simple_validator) - cookie = CookieAuth("session", cookie_validator) - multi = MultiAuth(bearer, cookie) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - # No Authorization header — falls through to cookie - response = client.get("/me", cookies={"session": VALID_COOKIE}) - assert response.status_code == 200 - assert response.json() == {"session": VALID_COOKIE} - - def test_no_source_matches_returns_401(self): - bearer = BearerTokenAuth(simple_validator) - cookie = CookieAuth("session", cookie_validator) - multi = MultiAuth(bearer, cookie) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me") - assert response.status_code == 401 - - def test_invalid_credential_does_not_fallthrough(self): - """If a credential is found but invalid, the next source is NOT tried.""" - second_called: list[bool] = [] - - async def tracking_validator(credential: str) -> dict: - second_called.append(True) - return {"from": "second"} - - bearer = BearerTokenAuth(simple_validator) # raises on wrong token - cookie = CookieAuth("session", tracking_validator) - multi = MultiAuth(bearer, cookie) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - # Bearer credential present but wrong — should NOT try cookie - response = client.get( - "/me", - headers={"Authorization": "Bearer wrong"}, - cookies={"session": VALID_COOKIE}, - ) - assert response.status_code == 401 - assert second_called == [] # cookie validator was never called - - def test_prefix_routes_to_correct_source(self): - """Prefix-based dispatch: only the matching source's validator is called.""" - user_calls: list[str] = [] - org_calls: list[str] = [] - - async def user_validator(credential: str) -> dict: - user_calls.append(credential) - return {"type": "user", "id": credential} - - async def org_validator(credential: str) -> dict: - org_calls.append(credential) - return {"type": "org", "id": credential} - - user_bearer = BearerTokenAuth(user_validator, prefix="user_") - org_bearer = BearerTokenAuth(org_validator, prefix="org_") - multi = MultiAuth(user_bearer, org_bearer) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - - response = client.get("/me", headers={"Authorization": "Bearer user_alice"}) - assert response.status_code == 200 - assert response.json() == {"type": "user", "id": "user_alice"} - assert user_calls == ["user_alice"] - assert org_calls == [] - - user_calls.clear() - - response = client.get("/me", headers={"Authorization": "Bearer org_acme"}) - assert response.status_code == 200 - assert response.json() == {"type": "org", "id": "org_acme"} - assert user_calls == [] - assert org_calls == ["org_acme"] - - def test_require_returns_new_multi_auth(self): - from fastapi_toolsets.security.sources import MultiAuth as MultiAuthClass - - bearer = BearerTokenAuth(role_validator) - multi = MultiAuth(bearer) - derived = multi.require(role="admin") - assert isinstance(derived, MultiAuthClass) - assert derived is not multi - - def test_require_forwards_kwargs_to_sources(self): - """multi.require() propagates to all sources that support it.""" - bearer = BearerTokenAuth(role_validator) - multi = MultiAuth(bearer) - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(multi.require(role="admin"))): - return user - - client = TestClient(_app(setup)) - response = client.get( - "/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"} - ) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - def test_require_skips_sources_without_require(self): - """Sources without require() are passed through unchanged.""" - header_auth = _HeaderAuth(secret="s3cr3t") - multi = MultiAuth(header_auth) - derived = multi.require(role="admin") - assert derived._sources[0] is header_auth - - def test_require_does_not_mutate_original(self): - bearer = BearerTokenAuth(role_validator, role="user") - multi = MultiAuth(bearer) - multi.require(role="admin") - assert bearer._kwargs == {"role": "user"} - - def test_require_mixed_sources(self): - """require() applies to sources with require(), skips those without.""" - from typing import cast - - bearer = BearerTokenAuth(role_validator) - header_auth = _HeaderAuth(secret="s3cr3t") - multi = MultiAuth(bearer, header_auth) - derived = multi.require(role="admin") - # bearer got require() applied, header_auth passed through - assert cast(BearerTokenAuth, derived._sources[0])._kwargs == {"role": "admin"} - assert derived._sources[1] is header_auth - - -class TestRequire: - def test_bearer_require_forwards_kwargs(self): - """require() creates a new instance that passes merged kwargs to validator.""" - bearer = BearerTokenAuth(role_validator) - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(bearer.require(role="admin"))): - return user - - client = TestClient(_app(setup)) - response = client.get( - "/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"} - ) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - def test_bearer_require_overrides_existing_kwarg(self): - """require() kwargs override kwargs set at instantiation.""" - bearer = BearerTokenAuth(role_validator, role="user") - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(bearer.require(role="admin"))): - return user - - client = TestClient(_app(setup)) - response = client.get( - "/admin", headers={"Authorization": f"Bearer {VALID_TOKEN}"} - ) - assert response.status_code == 200 - assert response.json()["role"] == "admin" - - def test_bearer_require_preserves_prefix(self): - """require() keeps the prefix of the original instance.""" - bearer = BearerTokenAuth(role_validator, prefix="user_") - derived = bearer.require(role="admin") - assert derived._prefix == "user_" - - def test_bearer_require_does_not_mutate_original(self): - """require() returns a new instance — original kwargs are unchanged.""" - bearer = BearerTokenAuth(role_validator, role="user") - bearer.require(role="admin") - assert bearer._kwargs == {"role": "user"} - - def test_cookie_require_forwards_kwargs(self): - async def scoped_validator(value: str, *, scope: str) -> dict: - if value != VALID_COOKIE: - raise UnauthorizedError() - return {"session": value, "scope": scope} - - cookie = CookieAuth("session", scoped_validator) - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(cookie.require(scope="admin"))): - return user - - client = TestClient(_app(setup)) - response = client.get("/admin", cookies={"session": VALID_COOKIE}) - assert response.status_code == 200 - assert response.json() == {"session": VALID_COOKIE, "scope": "admin"} - - def test_cookie_require_preserves_name(self): - cookie = CookieAuth("session", cookie_validator) - derived = cookie.require(scope="admin") - assert derived._name == "session" - - def test_bearer_require_in_multi_auth(self): - """require() instances work seamlessly inside MultiAuth.""" - PREFIXED_TOKEN = f"user_{VALID_TOKEN}" - - async def prefixed_role_validator(credential: str, *, role: str) -> dict: - if credential != PREFIXED_TOKEN: - raise UnauthorizedError() - return {"user": "alice", "role": role} - - bearer = BearerTokenAuth(prefixed_role_validator, prefix="user_") - multi = MultiAuth(bearer.require(role="admin")) - - def setup(app: FastAPI): - @app.get("/admin") - async def admin(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - response = client.get( - "/admin", headers={"Authorization": f"Bearer {PREFIXED_TOKEN}"} - ) - assert response.status_code == 200 - assert response.json() == {"user": "alice", "role": "admin"} - - -class TestSyncValidators: - """Sync (non-async) validators — covers the sync path in _call_validator.""" - - def test_bearer_sync_validator(self): - def sync_validator(credential: str) -> dict: - if credential != VALID_TOKEN: - raise UnauthorizedError() - return {"user": "alice"} - - bearer = BearerTokenAuth(sync_validator) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(bearer)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - def test_sync_validator_via_authenticate(self): - """authenticate() with sync validator (MultiAuth path).""" - - def sync_validator(credential: str) -> dict: - if credential != VALID_TOKEN: - raise UnauthorizedError() - return {"user": "alice"} - - bearer = BearerTokenAuth(sync_validator) - multi = MultiAuth(bearer) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - -class TestCookieAuthSigned: - """CookieAuth with HMAC-SHA256 signed cookies (secret_key path).""" - - SECRET = "test-hmac-secret" - - def test_valid_signed_cookie_via_set_cookie(self): - """set_cookie signs the value; the signed cookie is verified on read.""" - from fastapi import Response - - # secure=False for test client which runs over plain HTTP - auth = CookieAuth( - "session", cookie_validator, secret_key=self.SECRET, secure=False - ) - - def setup(app: FastAPI): - @app.get("/login") - async def login(response: Response): - auth.set_cookie(response, VALID_COOKIE) - return {"ok": True} - - @app.get("/me") - async def me(user=Security(auth)): - return user - - with TestClient(_app(setup)) as client: - client.get("/login") - response = client.get("/me") - assert response.status_code == 200 - assert response.json() == {"session": VALID_COOKIE} - - def test_set_cookie_has_secure_flag_by_default(self): - """set_cookie includes Secure flag when secure=True (the default).""" - from starlette.responses import Response as StarletteResponse - - auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET) - response = StarletteResponse() - auth.set_cookie(response, "value") - assert "secure" in response.headers["set-cookie"].lower() - - def test_set_cookie_no_secure_flag_when_disabled(self): - """set_cookie omits Secure flag when secure=False (local dev).""" - from starlette.responses import Response as StarletteResponse - - auth = CookieAuth( - "session", cookie_validator, secret_key=self.SECRET, secure=False - ) - response = StarletteResponse() - auth.set_cookie(response, "value") - assert "secure" not in response.headers["set-cookie"].lower() - - def test_tampered_signature_returns_401(self): - """A cookie whose HMAC signature has been modified is rejected.""" - import base64 as _b64 - import json as _json - import time as _time - - auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - data = _b64.urlsafe_b64encode( - _json.dumps({"v": VALID_COOKIE, "exp": int(_time.time()) + 9999}).encode() - ).decode() - response = client.get("/me", cookies={"session": f"{data}.invalidsig"}) - assert response.status_code == 401 - - def test_expired_signed_cookie_returns_401(self): - """A signed cookie past its expiry timestamp is rejected.""" - import base64 as _b64 - import hashlib as _hashlib - import hmac as _hmac - import json as _json - import time as _time - - auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - data = _b64.urlsafe_b64encode( - _json.dumps({"v": VALID_COOKIE, "exp": int(_time.time()) - 1}).encode() - ).decode() - sig = _hmac.new( - self.SECRET.encode(), data.encode(), _hashlib.sha256 - ).hexdigest() - response = client.get("/me", cookies={"session": f"{data}.{sig}"}) - assert response.status_code == 401 - - def test_invalid_json_payload_returns_401(self): - """A signed cookie whose payload is not valid JSON is rejected.""" - import base64 as _b64 - import hashlib as _hashlib - import hmac as _hmac - - auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - data = _b64.urlsafe_b64encode(b"not-valid-json").decode() - sig = _hmac.new( - self.SECRET.encode(), data.encode(), _hashlib.sha256 - ).hexdigest() - response = client.get("/me", cookies={"session": f"{data}.{sig}"}) - assert response.status_code == 401 - - def test_malformed_cookie_no_dot_returns_401(self): - """A signed cookie without the dot separator is rejected.""" - auth = CookieAuth("session", cookie_validator, secret_key=self.SECRET) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", cookies={"session": "nodothere"}) - assert response.status_code == 401 - - def test_hmac_without_secret_key_raises(self): - """Calling _hmac on an instance without secret_key raises RuntimeError.""" - auth = CookieAuth("session", cookie_validator) - with pytest.raises(RuntimeError, match="secret_key"): - auth._hmac("data") - - def test_set_cookie_without_secret(self): - """set_cookie without secret_key writes the raw value.""" - from starlette.responses import Response as StarletteResponse - - auth = CookieAuth("session", cookie_validator) - response = StarletteResponse() - auth.set_cookie(response, "rawvalue") - assert "session=rawvalue" in response.headers["set-cookie"] - - def test_delete_cookie(self): - """delete_cookie produces a Set-Cookie header that clears the session.""" - from starlette.responses import Response as StarletteResponse - - auth = CookieAuth("session", cookie_validator) - response = StarletteResponse() - auth.delete_cookie(response) - assert "session" in response.headers["set-cookie"] - - -# Minimal concrete subclass used only in tests below. -class _HeaderAuth(AuthSource): - """Reads a custom X-Token header — no FastAPI security scheme.""" - - def __init__(self, secret: str) -> None: - super().__init__() - self._secret = secret - - async def extract(self, request) -> str | None: - return request.headers.get("X-Token") or None - - async def authenticate(self, credential: str) -> dict: - if credential != self._secret: - raise UnauthorizedError() - return {"token": credential} - - -class TestAuthSource: - def test_cannot_instantiate_abstract_class(self): - with pytest.raises(TypeError): - AuthSource() - - def test_builtin_classes_are_auth_sources(self): - bearer = BearerTokenAuth(simple_validator) - cookie = CookieAuth("session", cookie_validator) - api_key = APIKeyHeaderAuth("X-API-Key", simple_validator) - assert isinstance(bearer, AuthSource) - assert isinstance(cookie, AuthSource) - assert isinstance(api_key, AuthSource) - - def test_custom_source_standalone_valid(self): - """Default __call__ wires extract + authenticate via Request injection.""" - auth = _HeaderAuth(secret="s3cr3t") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"X-Token": "s3cr3t"}) - assert response.status_code == 200 - assert response.json() == {"token": "s3cr3t"} - - def test_custom_source_standalone_missing_credential(self): - auth = _HeaderAuth(secret="s3cr3t") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me") # no X-Token header - assert response.status_code == 401 - - def test_custom_source_standalone_invalid_credential(self): - auth = _HeaderAuth(secret="s3cr3t") - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(auth)): - return user - - client = TestClient(_app(setup)) - response = client.get("/me", headers={"X-Token": "wrong"}) - assert response.status_code == 401 - - def test_custom_source_in_multi_auth(self): - """Custom AuthSource works transparently inside MultiAuth.""" - header_auth = _HeaderAuth(secret="s3cr3t") - bearer = BearerTokenAuth(simple_validator) - multi = MultiAuth(bearer, header_auth) - - def setup(app: FastAPI): - @app.get("/me") - async def me(user=Security(multi)): - return user - - client = TestClient(_app(setup)) - - # Bearer matches first - response = client.get("/me", headers={"Authorization": f"Bearer {VALID_TOKEN}"}) - assert response.status_code == 200 - assert response.json() == {"user": "alice"} - - # No bearer → falls through to custom header source - response = client.get("/me", headers={"X-Token": "s3cr3t"}) - assert response.status_code == 200 - assert response.json() == {"token": "s3cr3t"} - - -def _make_async_client_mock(get_return=None, post_return=None): - """Return a patched httpx.AsyncClient context-manager mock.""" - mock_client = AsyncMock() - if get_return is not None: - mock_client.get.return_value = get_return - if post_return is not None: - mock_client.post.return_value = post_return - cm = MagicMock() - cm.__aenter__ = AsyncMock(return_value=mock_client) - cm.__aexit__ = AsyncMock(return_value=None) - return cm, mock_client - - -class TestEncodeDecodeOAuthState: - def test_encode_returns_base64url_string(self): - 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" - state_token = "test-state-token" - assert ( - oauth_decode_state( - 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_state_token="any", fallback="/home") - == "/home" - ) - - def test_decode_null_string_returns_fallback(self): - assert ( - 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_state_token="any", fallback="/home" - ) - == "/home" - ) - - def test_decode_handles_missing_padding(self): - url = "https://example.com/x" - state_token = "test-state-token" - encoded = oauth_encode_state(url, state_token).rstrip("=") - assert ( - 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_state_token_is_random(self): - assert oauth_generate_state_token() != oauth_generate_state_token() - - -class TestBuildAuthorizationRedirect: - def test_returns_redirect_response(self): - from fastapi.responses import RedirectResponse - - 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", - state_token="test-state-token", - ) - assert isinstance(response, RedirectResponse) - - def test_redirect_location_contains_all_params(self): - 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", - state_token=state_token, - ) - location = response.headers["location"] - parsed = urlparse(location) - assert ( - parsed.scheme + "://" + parsed.netloc + parsed.path - == "https://auth.example.com/authorize" - ) - params = parse_qs(parsed.query) - assert params["client_id"] == ["my-client"] - assert params["response_type"] == ["code"] - assert params["scope"] == ["openid email"] - assert params["redirect_uri"] == ["https://app.example.com/callback"] - assert ( - oauth_decode_state( - params["state"][0], expected_state_token=state_token, fallback="" - ) - == "https://app.example.com/dashboard" - ) - - -class TestResolveProviderUrls: - def _discovery(self, *, userinfo=True): - doc = { - "authorization_endpoint": "https://auth.example.com/authorize", - "token_endpoint": "https://auth.example.com/token", - } - if userinfo: - doc["userinfo_endpoint"] = "https://auth.example.com/userinfo" - return doc - - @pytest.mark.anyio - async def test_returns_all_endpoints(self): - mock_resp = MagicMock() - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = self._discovery() - cm, mock_client = _make_async_client_mock(get_return=mock_resp) - - oauth_resolve_provider_urls.cache_clear() - with patch("httpx.AsyncClient", return_value=cm): - auth_url, token_url, userinfo_url = await oauth_resolve_provider_urls( - "https://auth.example.com/.well-known/openid-configuration" - ) - - assert auth_url == "https://auth.example.com/authorize" - assert token_url == "https://auth.example.com/token" - assert userinfo_url == "https://auth.example.com/userinfo" - - @pytest.mark.anyio - async def test_userinfo_url_none_when_absent(self): - mock_resp = MagicMock() - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = self._discovery(userinfo=False) - cm, mock_client = _make_async_client_mock(get_return=mock_resp) - - oauth_resolve_provider_urls.cache_clear() - with patch("httpx.AsyncClient", return_value=cm): - _, _, userinfo_url = await oauth_resolve_provider_urls( - "https://auth.example.com/.well-known/openid-configuration" - ) - - assert userinfo_url is None - - @pytest.mark.anyio - async def test_caches_discovery_document(self): - mock_resp = MagicMock() - mock_resp.raise_for_status = MagicMock() - mock_resp.json.return_value = self._discovery() - cm, mock_client = _make_async_client_mock(get_return=mock_resp) - - url = "https://auth.example.com/.well-known/openid-configuration" - oauth_resolve_provider_urls.cache_clear() - with patch("httpx.AsyncClient", return_value=cm): - await oauth_resolve_provider_urls(url) - await oauth_resolve_provider_urls(url) - - assert mock_client.get.call_count == 1 - - -class TestFetchUserinfo: - @pytest.mark.anyio - async def test_returns_userinfo_payload(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = {"access_token": "tok123"} - - userinfo_resp = MagicMock() - userinfo_resp.raise_for_status = MagicMock() - userinfo_resp.json.return_value = { - "sub": "user-1", - "email": "alice@example.com", - } - - cm, mock_client = _make_async_client_mock( - post_return=token_resp, get_return=userinfo_resp - ) - - with patch("httpx.AsyncClient", return_value=cm): - result = await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="client-id", - client_secret="client-secret", - redirect_uri="https://app.example.com/callback", - ) - - assert result == {"sub": "user-1", "email": "alice@example.com"} - - @pytest.mark.anyio - async def test_posts_correct_token_request_and_uses_bearer(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = {"access_token": "tok123"} - - userinfo_resp = MagicMock() - userinfo_resp.raise_for_status = MagicMock() - userinfo_resp.json.return_value = {} - - cm, mock_client = _make_async_client_mock( - post_return=token_resp, get_return=userinfo_resp - ) - - with patch("httpx.AsyncClient", return_value=cm): - await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="my-client", - client_secret="my-secret", - redirect_uri="https://app.example.com/callback", - ) - - mock_client.post.assert_called_once_with( - "https://auth.example.com/token", - data={ - "grant_type": "authorization_code", - "code": "authcode123", - "client_id": "my-client", - "client_secret": "my-secret", - "redirect_uri": "https://app.example.com/callback", - }, - headers={"Accept": "application/json"}, - ) - mock_client.get.assert_called_once_with( - "https://auth.example.com/userinfo", - headers={"Authorization": "Bearer tok123"}, - ) - - @pytest.mark.anyio - async def test_raises_on_unsupported_token_type(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = {"access_token": "tok123", "token_type": "mac"} - - cm, _ = _make_async_client_mock(post_return=token_resp, get_return=MagicMock()) - - with patch("httpx.AsyncClient", return_value=cm): - with pytest.raises(ValueError, match="unsupported token_type"): - await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="client-id", - client_secret="client-secret", - redirect_uri="https://app.example.com/callback", - ) - - @pytest.mark.anyio - async def test_accepts_bearer_token_type_case_insensitive(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = { - "access_token": "tok123", - "token_type": "Bearer", - } - - userinfo_resp = MagicMock() - userinfo_resp.raise_for_status = MagicMock() - userinfo_resp.json.return_value = {"sub": "user-1"} - - cm, _ = _make_async_client_mock( - post_return=token_resp, get_return=userinfo_resp - ) - - with patch("httpx.AsyncClient", return_value=cm): - result = await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="client-id", - client_secret="client-secret", - redirect_uri="https://app.example.com/callback", - ) - assert result == {"sub": "user-1"} - - @pytest.mark.anyio - async def test_raises_when_required_scopes_not_granted(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = {"access_token": "tok123", "scope": "openid"} - - cm, _ = _make_async_client_mock(post_return=token_resp, get_return=MagicMock()) - - with patch("httpx.AsyncClient", return_value=cm): - with pytest.raises(ValueError, match="required scopes"): - await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="client-id", - client_secret="client-secret", - redirect_uri="https://app.example.com/callback", - required_scopes="openid email profile", - ) - - @pytest.mark.anyio - async def test_passes_when_all_required_scopes_granted(self): - token_resp = MagicMock() - token_resp.raise_for_status = MagicMock() - token_resp.json.return_value = { - "access_token": "tok123", - "scope": "openid email profile", - } - - userinfo_resp = MagicMock() - userinfo_resp.raise_for_status = MagicMock() - userinfo_resp.json.return_value = {"sub": "user-1", "email": "a@b.com"} - - cm, _ = _make_async_client_mock( - post_return=token_resp, get_return=userinfo_resp - ) - - with patch("httpx.AsyncClient", return_value=cm): - result = await oauth_fetch_userinfo( - token_url="https://auth.example.com/token", - userinfo_url="https://auth.example.com/userinfo", - code="authcode123", - client_id="client-id", - client_secret="client-secret", - redirect_uri="https://app.example.com/callback", - required_scopes="openid email", - ) - assert result["email"] == "a@b.com" diff --git a/uv.lock b/uv.lock index f3bc558c..5717c76a 100644 --- a/uv.lock +++ b/uv.lock @@ -341,7 +341,6 @@ dependencies = [ [package.optional-dependencies] all = [ - { name = "async-lru" }, { name = "httpx" }, { name = "prometheus-client" }, { name = "pytest" }, @@ -359,10 +358,6 @@ pytest = [ { name = "pytest" }, { name = "pytest-xdist" }, ] -security = [ - { name = "async-lru" }, - { name = "httpx" }, -] [package.dev-dependencies] dev = [ @@ -402,12 +397,10 @@ tests = [ [package.metadata] requires-dist = [ - { name = "async-lru", marker = "extra == 'security'", specifier = ">=1.0" }, { name = "asyncpg", specifier = ">=0.29.0" }, { name = "fastapi", specifier = ">=0.100.0" }, - { name = "fastapi-toolsets", extras = ["cli", "metrics", "pytest", "security"], marker = "extra == 'all'" }, + { name = "fastapi-toolsets", extras = ["cli", "metrics", "pytest"], marker = "extra == 'all'" }, { name = "httpx", marker = "extra == 'pytest'", specifier = ">=0.25.0" }, - { name = "httpx", marker = "extra == 'security'", specifier = ">=0.25.0" }, { name = "prometheus-client", marker = "extra == 'metrics'", specifier = ">=0.20.0" }, { name = "pydantic", specifier = ">=2.0" }, { name = "pytest", marker = "extra == 'pytest'", specifier = ">=8.0.0" }, @@ -415,7 +408,7 @@ requires-dist = [ { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0" }, { name = "typer", marker = "extra == 'cli'", specifier = ">=0.9.0" }, ] -provides-extras = ["cli", "metrics", "security", "pytest", "all"] +provides-extras = ["cli", "metrics", "pytest", "all"] [package.metadata.requires-dev] dev = [ From 5f1ab8275ba31e83fed2f0b115811a4df2621cdd Mon Sep 17 00:00:00 2001 From: d3vyce Date: Tue, 30 Jun 2026 16:16:37 -0400 Subject: [PATCH 2/3] docs: add v5 migration guide --- docs/migration/v4.md | 4 +- docs/migration/v5.md | 161 +++++++++++++++++++++++++++++++++++++++++++ docs/module/cli.md | 4 +- zensical.toml | 3 +- 4 files changed, 166 insertions(+), 6 deletions(-) create mode 100644 docs/migration/v5.md diff --git a/docs/migration/v4.md b/docs/migration/v4.md index fc7587e7..aeaf656d 100644 --- a/docs/migration/v4.md +++ b/docs/migration/v4.md @@ -21,7 +21,7 @@ The function creates and manages its own **dedicated session** internally, yield user.balance += 100 # With a custom lock mode - async with lock_tables(session, [Order], mode=LockMode.EXCLUSIVE): + async with lock_tables(session=session, tables=[Order], mode=LockMode.EXCLUSIVE): await process_order(session, order_id) ``` @@ -35,6 +35,6 @@ The function creates and manages its own **dedicated session** internally, yield user.balance += 100 # With a custom lock mode - async with lock_tables(session_maker, [Order], mode=LockMode.EXCLUSIVE) as session: + async with lock_tables(session_maker=session_maker, tables=[Order], mode=LockMode.EXCLUSIVE) as session: await process_order(session, order_id) ``` diff --git a/docs/migration/v5.md b/docs/migration/v5.md new file mode 100644 index 00000000..2d0d01dd --- /dev/null +++ b/docs/migration/v5.md @@ -0,0 +1,161 @@ +# Migrating to v5.0 + +This page covers every breaking change introduced in **v5.0** and the steps required to update your code. + +--- + +## Database + +`db.py` is now the `db/` package, built around one object, [`Database`](../reference/db.md#fastapi_toolsets.db.Database), that owns the engine and sessionmaker. The free functions that took a `session_maker` you built and passed around yourself are gone from request-handling code; `Database` builds the sessionmaker for you. + +### `create_db_dependency` / `create_db_context` removed in favor of `Database` + +Build one `Database` with your URL (or an existing `engine=`), then use the instance directly as the FastAPI dependency, and `db.session()` for sessions outside request handlers. + +=== "Before (`v4`)" + + ```python + from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker + from fastapi_toolsets.db import create_db_dependency, create_db_context + + engine = create_async_engine("postgresql+asyncpg://...") + SessionLocal = async_sessionmaker(engine, expire_on_commit=False) + + get_db = create_db_dependency(session_maker=SessionLocal) + get_db_context = create_db_context(session_maker=SessionLocal) + + @app.get("/users") + async def list_users(session: AsyncSession = Depends(get_db)): + ... + + async def seed(): + async with get_db_context() as session: + ... + ``` + +=== "Now (`v5`)" + + ```python + from fastapi_toolsets.db import Database + + db = Database(url="postgresql+asyncpg://...") + + @app.get("/users") + async def list_users(session: AsyncSession = Depends(db)): + ... + + async def seed(): + async with db.session() as session: + ... + ``` + +Call `db.install(app)` to also commit before the response is sent (instead of in dependency teardown) and to dispose the engine on shutdown. See [the db module docs](../module/db.md#committing-before-the-response). + +### `get_transaction` renamed to `transaction` + +Same behavior (savepoint when already in a transaction, new transaction otherwise), new name, same import path. + +=== "Before (`v4`)" + + ```python + from fastapi_toolsets.db import get_transaction + + async with get_transaction(session=session): + session.add(model) + ``` + +=== "Now (`v5`)" + + ```python + from fastapi_toolsets.db import transaction + + async with transaction(session=session): + session.add(model) + ``` + +If you have a `Database` instance, `db.begin()` opens a session already inside a transaction: + +```python +async with db.begin() as session: + session.add(User(name="ada")) +``` + +### `lock_tables` is now also a `Database` method + +The free `lock_tables(session_maker, tables, ...)` function still exists for callers who manage their own session factory, but prefer `db.lock_tables(tables, ...)`, which drops the `session_maker` argument: + +=== "Before (`v4`)" + + ```python + from fastapi_toolsets.db import lock_tables, LockMode + + async with lock_tables(session_maker=session_maker, tables=[Order], mode=LockMode.EXCLUSIVE) as session: + await process_order(session, order_id) + ``` + +=== "Now (`v5`)" + + ```python + from fastapi_toolsets.db import LockMode + + async with db.lock_tables(tables=[Order], mode=LockMode.EXCLUSIVE) as session: + await process_order(session, order_id) + ``` + +### `create_database` and `cleanup_tables` moved to `fastapi_toolsets.db.testing` + +=== "Before (`v4`)" + + ```python + from fastapi_toolsets.db import create_database, cleanup_tables + ``` + +=== "Now (`v5`)" + + ```python + from fastapi_toolsets.db.testing import create_database, cleanup_tables + ``` + +## Fixtures + +### `get_obj_by_attr` / `get_field_by_attr` are now `FixtureRegistry` methods + +Both also change their first argument: instead of the fixture *function*, pass the fixture's *registered name* and let the registry look it up. + +=== "Before (`v4`)" + + ```python + from fastapi_toolsets.fixtures import get_obj_by_attr, get_field_by_attr + + @fixtures.register(depends_on=["roles"]) + def users(): + admin_role = get_obj_by_attr(fixtures=roles, attr_name="name", value="admin") + admin_role_id = get_field_by_attr(fixtures=roles, attr_name="name", value="admin") + return [User(id=1, username="alice", role_id=admin_role.id)] + ``` + +=== "Now (`v5`)" + + ```python + @fixtures.register(depends_on=["roles"]) + def users(): + admin_role = fixtures.obj(name="roles", attr_name="name", value="admin") + admin_role_id = fixtures.field(name="roles", attr_name="name", value="admin") + return [User(id=1, username="alice", role_id=admin_role.id)] + ``` + +### Loading/listing by context now always includes `Context.BASE` + +`FixtureRegistry.get_variants`/`get_by_context`, `load_fixtures_by_context`, and the `fixtures list`/`fixtures load` CLI commands now implicitly union every queried context with `Context.BASE`. In `v4`, requesting a single context (e.g. `"testing"`) returned only fixtures tagged with that context; in `v5` it also returns every `Context.BASE`-tagged fixture. + +If you relied on strict exclusion of base fixtures, move them out of `Context.BASE` into their own context. + +### `load_fixtures` / `load_fixtures_by_context` now return reloaded instances + +Loaded objects are re-queried from the database after insert, with all relationships eager-loaded, instead of returning the exact objects your fixture function constructed. Code that compares returned objects by identity to the fixture function's output, or that expected relationships to remain unloaded, should be updated — and expect one extra query per model type per load. + +## Security + +The security module has been removed and moved to a dedicated python package: [`fastapi-multiauth`](https://github.com/d3vyce/fastapi-multiauth). + +Run `uv add fastapi-multiauth` and replace `from fastapi_toolsets.security import ...` with `from fastapi_multiauth import ...`. diff --git a/docs/module/cli.md b/docs/module/cli.md index a3933c3c..f31da978 100644 --- a/docs/module/cli.md +++ b/docs/module/cli.md @@ -24,7 +24,7 @@ Configure the CLI in your `pyproject.toml`: ```toml [tool.fastapi-toolsets] -cli = "myapp.cli:cli" # Custom Typer app +custom_cli = "myapp.cli:cli" # Custom Typer app fixtures = "myapp.fixtures:registry" # FixtureRegistry instance db_context = "myapp.db:db_context" # Async context manager for sessions ``` @@ -99,7 +99,7 @@ def hello(): ```toml [tool.fastapi-toolsets] -cli = "myapp.cli:cli" +custom_cli = "myapp.cli:cli" ``` --- diff --git a/zensical.toml b/zensical.toml index d4cc6fe6..5b501e74 100644 --- a/zensical.toml +++ b/zensical.toml @@ -121,7 +121,6 @@ Modules = [ {Models = "module/models.md"}, {Pytest = "module/pytest.md"}, {Schemas = "module/schemas.md"}, - {Security = "module/security.md"}, ] [[project.nav]] @@ -137,7 +136,6 @@ Reference = [ {Models = "reference/models.md"}, {Pytest = "reference/pytest.md"}, {Schemas = "reference/schemas.md"}, - {Security = "reference/security.md"}, ] [[project.nav]] @@ -147,6 +145,7 @@ Examples = [ [[project.nav]] Migration = [ + {"v5.0" = "migration/v5.md"}, {"v4.0" = "migration/v4.md"}, {"v3.0" = "migration/v3.md"}, {"v2.0" = "migration/v2.md"}, From 00368938ef984ad3a25b91c7048d5e0997fc29b1 Mon Sep 17 00:00:00 2001 From: d3vyce Date: Sat, 11 Jul 2026 09:06:31 -0400 Subject: [PATCH 3/3] Version 5.0.0 --- pyproject.toml | 2 +- src/fastapi_toolsets/__init__.py | 2 +- uv.lock | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 3806f625..566dfbfd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fastapi-toolsets" -version = "5.0.0b2" +version = "5.0.0" description = "Production-ready utilities for FastAPI applications" readme = "README.md" license = "MIT" diff --git a/src/fastapi_toolsets/__init__.py b/src/fastapi_toolsets/__init__.py index d1caf290..df8d8df9 100644 --- a/src/fastapi_toolsets/__init__.py +++ b/src/fastapi_toolsets/__init__.py @@ -24,4 +24,4 @@ async def get_user(user_id: int, session = Depends(db)): return Response(data={"user": user.username}, message="Success") """ -__version__ = "5.0.0b2" +__version__ = "5.0.0" diff --git a/uv.lock b/uv.lock index 5717c76a..26b52514 100644 --- a/uv.lock +++ b/uv.lock @@ -330,7 +330,7 @@ wheels = [ [[package]] name = "fastapi-toolsets" -version = "5.0.0b2" +version = "5.0.0" source = { editable = "." } dependencies = [ { name = "asyncpg" },