wip

src/fastapi_toolsets/security/sources/bearer.py

@@ -33,23 +33,6 @@ class BearerTokenAuth(AuthSource):
             for user tokens, ``"org_"`` for org tokens).
         **kwargs: Extra keyword arguments forwarded to the validator on every
             call (e.g. ``role=Role.ADMIN``).
-
-    Example::
-
-        async def verify_token(token: str, *, role: Role) -> User:
-            user = await db.get_by_token(token)   # token includes prefix
-            if not user or user.role != role:
-                raise UnauthorizedError()
-            return user
-
-        bearer_admin = BearerTokenAuth(verify_token, prefix="user_", role=Role.ADMIN)
-
-        # Generate a token to store in DB and return to the client:
-        token = bearer_admin.generate_token()   # e.g. "user_Xk3..."
-
-        @app.get("/admin")
-        async def admin_route(user: User = Security(bearer_admin)):
-            return user
     """
 
     def __init__(
@@ -64,18 +47,12 @@ class BearerTokenAuth(AuthSource):
         self._kwargs = kwargs
         self._scheme = HTTPBearer(auto_error=False)
 
-        # Capture locals for the closure — self._scheme cannot be referenced
-        # inside the Annotated default because annotations are evaluated at
-        # function-definition time (no `from __future__ import annotations`).
         _scheme = self._scheme
         _validator = validator
         _kwargs = kwargs
         _prefix = prefix
 
         async def _call(
-            # security_scopes is unused in the body but its presence in the
-            # signature tells FastAPI to aggregate scopes from Security() calls
-            # up the dependency chain and expose them in the OpenAPI schema.
             security_scopes: SecurityScopes,  # noqa: ARG001
             credentials: Annotated[
                 HTTPAuthorizationCredentials | None, Depends(_scheme)
@@ -88,9 +65,6 @@ class BearerTokenAuth(AuthSource):
                 raise UnauthorizedError()
             return await _validator(token, **_kwargs)
 
-        # __call__ must be defined on the class (not the instance) so that
-        # callable(self) returns True. We expose the closure's signature via
-        # __signature__ so FastAPI resolves the correct sub-dependencies.
         self._call_fn = _call
         self.__signature__ = inspect.signature(_call)
 
@@ -123,20 +97,7 @@ class BearerTokenAuth(AuthSource):
         return await self._validator(credential, **self._kwargs)
 
     def require(self, **kwargs: Any) -> "BearerTokenAuth":
-        """Return a new instance with additional (or overriding) validator kwargs.
+        """Return a new instance with additional (or overriding) validator kwargs."""
-
-        Useful for specifying per-endpoint requirements inline without
-        declaring a new top-level variable::
-
-            bearer = BearerTokenAuth(verify_token)
-
-            @app.get("/admin")
-            async def admin(user: User = Security(bearer.require(role=Role.ADMIN))):
-                return user
-
-        The ``prefix`` is preserved. New kwargs are merged over existing ones
-        (new values win on conflict).
-        """
         return BearerTokenAuth(
             self._validator,
             prefix=self._prefix,
@@ -157,13 +118,6 @@ class BearerTokenAuth(AuthSource):
 
         Returns:
             A ready-to-use token string (e.g. ``"user_Xk3..."``).
-
-        Example::
-
-            bearer = BearerTokenAuth(verify_token, prefix="user_")
-            token = bearer.generate_token()   # "user_<random>"
-            await db.store_token(user_id, token)
-            return {"access_token": token, "token_type": "bearer"}
         """
         token = secrets.token_urlsafe(nbytes)
         if self._prefix is not None:

src/fastapi_toolsets/security/sources/cookie.py

@@ -1,9 +1,14 @@
 """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
+from fastapi import Depends, Request, Response
 from fastapi.security import APIKeyCookie, SecurityScopes
 
 from fastapi_toolsets.exceptions import UnauthorizedError
@@ -15,42 +20,43 @@ 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 to read the credential from.
+        name: Cookie name.
-        validator: Async callable that receives the cookie value and any extra
+        validator: 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.
         **kwargs: Extra keyword arguments forwarded to the validator on every
-            call.
+            call (e.g. ``role=Role.ADMIN``).
-
-    Example::
-
-        async def verify_session(session_id: str) -> User:
-            user = await db.get_by_session(session_id)
-            if not user:
-                raise UnauthorizedError()
-            return user
-
-        cookie_auth = CookieAuth("session", verify_session)
-
-        @app.get("/me")
-        async def me(user: User = Security(cookie_auth)):
-            return user
     """
 
     def __init__(
         self,
         name: str,
         validator: Callable[..., Any],
+        *,
+        secret_key: str | None = None,
+        ttl: int = 86400,
         **kwargs: Any,
     ) -> None:
         self._name = name
         self._validator = validator
+        self._secret_key = secret_key
+        self._ttl = ttl
         self._kwargs = kwargs
         self._scheme = APIKeyCookie(name=name, auto_error=False)
 
         _scheme = self._scheme
-        _validator = validator
+        _self = self
         _kwargs = kwargs
 
         async def _call(
@@ -59,7 +65,8 @@ class CookieAuth(AuthSource):
         ) -> Any:
             if value is None:
                 raise UnauthorizedError()
-            return await _validator(value, **_kwargs)
+            plain = _self._verify(value)
+            return await _self._validator(plain, **_kwargs)
 
         self._call_fn = _call
         self.__signature__ = inspect.signature(_call)
@@ -67,28 +74,71 @@ class CookieAuth(AuthSource):
     async def __call__(self, **kwargs: Any) -> Any:
         return await self._call_fn(**kwargs)
 
+    def _hmac(self, data: str) -> str:
+        assert self._secret_key is not None
+        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:
-        """Extract the cookie value from the request without validating."""
         return request.cookies.get(self._name)
 
     async def authenticate(self, credential: str) -> Any:
-        """Validate a credential and return the identity."""
+        plain = self._verify(credential)
-        return await self._validator(credential, **self._kwargs)
+        return await self._validator(plain, **self._kwargs)
 
     def require(self, **kwargs: Any) -> "CookieAuth":
-        """Return a new instance with additional (or overriding) validator kwargs.
+        """Return a new instance with additional (or overriding) validator kwargs."""
-
-        The cookie name is preserved. New kwargs are merged over existing ones
-        (new values win on conflict)::
-
-            cookie = CookieAuth("session", verify_session)
-
-            @app.get("/admin")
-            async def admin(user: User = Security(cookie.require(role=Role.ADMIN))):
-                return user
-        """
         return CookieAuth(
             self._name,
             self._validator,
+            secret_key=self._secret_key,
+            ttl=self._ttl,
             **{**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",
+            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")

src/fastapi_toolsets/security/sources/oauth2.py

@@ -23,17 +23,6 @@ class OAuth2Auth(AuthSource):
             arguments, and returns the authenticated identity.
         **kwargs: Extra keyword arguments forwarded to the validator on every
             call.
-
-    Example::
-
-        async def verify_token(token: str) -> User:
-            ...
-
-        oauth2_auth = OAuth2Auth(token_url="/token", validator=verify_token)
-
-        @app.get("/me")
-        async def me(user: User = Security(oauth2_auth)):
-            return user
     """
 
     def __init__(
@@ -78,17 +67,7 @@ class OAuth2Auth(AuthSource):
         return await self._validator(credential, **self._kwargs)
 
     def require(self, **kwargs: Any) -> "OAuth2Auth":
-        """Return a new instance with additional (or overriding) validator kwargs.
+        """Return a new instance with additional (or overriding) validator kwargs."""
-
-        The token URL is preserved. New kwargs are merged over existing ones
-        (new values win on conflict)::
-
-            oauth2 = OAuth2Auth("/token", verify_token)
-
-            @app.get("/admin")
-            async def admin(user: User = Security(oauth2.require(role=Role.ADMIN))):
-                return user
-        """
         return OAuth2Auth(
             self._token_url,
             self._validator,

src/fastapi_toolsets/security/sources/openid.py

@@ -29,33 +29,6 @@ class OpenIDAuth(AuthSource):
             on failure.
         **kwargs: Extra keyword arguments forwarded to the validator on every
             call (e.g. ``audience="my-app"``).
-
-    Example — Google::
-
-        import jwt  # e.g. PyJWT or python-jose
-
-        async def verify_google_token(token: str, *, audience: str) -> User:
-            payload = jwt.decode(token, google_public_keys, algorithms=["RS256"],
-                                 audience=audience)
-            return User(email=payload["email"], name=payload["name"])
-
-        google_auth = OpenIDAuth(
-            "https://accounts.google.com/.well-known/openid-configuration",
-            verify_google_token,
-            audience="my-client-id",
-        )
-
-        @app.get("/me")
-        async def me(user: User = Security(google_auth)):
-            return user
-
-    Multiple providers with :func:`~fastapi_toolsets.security.MultiAuth`::
-
-        multi = MultiAuth(google_auth, github_auth)
-
-        @app.get("/data")
-        async def data(user: User = Security(multi)):
-            return user
     """
 
     def __init__(
@@ -77,9 +50,6 @@ class OpenIDAuth(AuthSource):
 
         async def _call(
             security_scopes: SecurityScopes,  # noqa: ARG001
-            # OpenIdConnect (OAuth2 base) returns the full Authorization header
-            # value (e.g. "Bearer <token>"), unlike OAuth2PasswordBearer which
-            # strips the scheme prefix.
             authorization: Annotated[str | None, Depends(_scheme)] = None,
         ) -> Any:
             if authorization is None:
@@ -109,17 +79,7 @@ class OpenIDAuth(AuthSource):
         return await self._validator(credential, **self._kwargs)
 
     def require(self, **kwargs: Any) -> "OpenIDAuth":
-        """Return a new instance with additional (or overriding) validator kwargs.
+        """Return a new instance with additional (or overriding) validator kwargs."""
-
-        The discovery URL is preserved. New kwargs are merged over existing ones
-        (new values win on conflict)::
-
-            google_auth = OpenIDAuth(discovery_url, verify_google_token, audience="app")
-
-            @app.get("/admin")
-            async def admin(user: User = Security(google_auth.require(role=Role.ADMIN))):
-                return user
-        """
         return OpenIDAuth(
             self._openid_connect_url,
             self._validator,