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/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..3daf397b --- /dev/null +++ b/docs/migration/v5.md @@ -0,0 +1,147 @@ +# 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 + +=== "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(roles, "name", "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("roles", "name", "admin") + return [User(id=1, username="alice", role_id=admin_role.id)] + ``` + +## 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 c80a7367..43d5b2ab 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 ``` 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 9ed885e1..2da26386 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 = [ 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"},