Updated July 1, 2026

On this page

Reference for fastware[auth]: JWT tokens, bcrypt hashing, user stores, get_current_user/require_role DI, CSRF middleware, sessions, and rate limiting.

#Auth API Reference

Warning

The auth module requires the fastware[auth] extra, which installs PyJWT and bcrypt. Install with:

bash

uv add "fastware[auth]"

Importing fastware.auth without these dependencies will raise ImportError at call time.

The auth module provides JWT token operations, password hashing, user storage, CSRF protection, rate limiting, and session cookie management. All functions are pure and DI-compatible, with no framework-specific dependencies beyond fastware's own ASGI types.

#`src.fastware.auth`

Authentication module providing JWT token creation and verification, bcrypt password hashing, user storage, CSRF protection, and rate limiting.

Pure functions and DI-compatible factories with no framework-specific dependencies beyond fastware's own asgi types. Keeps auth logic testable and reusable.

#`create_token`

python

def create_token(username: str, role: str, secret: str, expires_hours: int=720) -> str

Create a signed JWT with sub, role, exp and iat claims (HS256).

#`verify_token`

python

def verify_token(token: str, secret: str) -> dict[str, Any] | None

Decode and validate a JWT. Returns claims dict or None if invalid.

Only token-validation failures map to None; programming errors (e.g. a None secret) propagate.

#`hash_password`

python

def hash_password(plain: str) -> str

Hash a plaintext password with bcrypt.

Raises ValueError for passwords longer than 72 bytes (UTF-8): bcrypt ignores everything past byte 72, so accepting them would silently weaken the password.

#`verify_password`

python

def verify_password(plain: str, hashed: str) -> bool

Check a plaintext password against a bcrypt hash.

#`UserStore`

Abstract user storage interface.

Subclasses overriding __init__ must call super().__init__() so the read-modify-write lock is set up.

#`load_users`

python

def load_users(self) -> list[dict[str, str]]

#`save_users`

python

def save_users(self, users: list[dict[str, str]]) -> None

#`find_user`

python

def find_user(self, username: str) -> dict[str, str] | None

#`create_user`

python

def create_user(self, username: str, password: str, role: str) -> dict[str, str]

Create a new user. Raises ValueError if username already exists.

#`delete_user`

python

def delete_user(self, username: str) -> None

Delete a user by username. Raises LookupError if not found.

#`JSONFileUserStore`

User storage backed by a JSON file.

#`load_users`

python

def load_users(self) -> list[dict[str, str]]

#`save_users`

python

def save_users(self, users: list[dict[str, str]]) -> None

Atomically write the user list (temp file + os.replace).

A crash mid-write can never truncate or corrupt the existing file.

#`get_current_user`

python

def get_current_user(request: Any, *, allow_query_token: bool=False) -> dict[str, Any]

Extract and validate JWT from Authorization header or session cookie.

Token resolution order:

Authorization: Bearer header
session cookie
?token= query parameter -- only if allow_query_token=True.

Off by default because query strings leak into access logs, proxies, and browser history. Opt in via a wrapper dep: deps={"user": lambda request: get_current_user(request, allow_query_token=True)}

Reads the JWT secret from request.state["config"]["jwt_secret"]. Returns decoded claims dict. Raises HTTPError(401) on failure.

#`require_role`

python

def require_role(role: str) -> Callable

Return a DI factory that checks the current user has the given role.

Usage: @router.get("/admin", deps={"user": require_role("admin")})

#`CSRFMiddleware`

Double-submit cookie CSRF protection (pure ASGI).

For state-changing requests (POST, PUT, PATCH, DELETE) that aren't exempt, validates that: 1. A csrf_token cookie is present. 2. An X-CSRF-Token header is present. 3. The two values match.

Constructor args: app: inner ASGI application exempt_paths: list of path prefixes to skip CSRF checks disabled: bypass all checks (for testing)

#`set_session_cookies`

python

def set_session_cookies(token: str, csrf_token: str) -> list[str]

Build Set-Cookie header strings for session and CSRF cookies.

Returns a list of two Set-Cookie strings:

session: httponly, samesite=lax (not readable by JS)
csrf_token: js-readable (no httponly), samesite=lax

#`clear_session_cookies`

python

def clear_session_cookies() -> list[str]

Build Set-Cookie header strings that clear session and CSRF cookies.

#`rate_limit`

python

def rate_limit(rate: str, key_func: Callable | None=None) -> Callable

Decorator for per-client rate limiting using a token bucket.

Usage: @router.get("/api/search") @rate_limit("5/minute") async def search(request): ...

Args:

rate: Rate string like "5/minute", "10/second", "100/hour".
key_func: Optional callable(request) -> str for custom bucket keys.

Defaults to client IP from ASGI scope.

python

class TestJWTTokens:
    """create_token and verify_token."""

    def test_create_and_verify(self):
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret)
        claims = verify_token(token, secret)
        assert claims is not None
        assert claims["sub"] == "alice"
        assert claims["role"] == "admin"
        assert "iat" in claims
        assert "exp" in claims

    def test_expired_token_returns_none(self):
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret, expires_hours=-1)
        assert verify_token(token, secret) is None

    def test_tampered_token_returns_none(self):
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret)
        # Flip a character in the middle of the signature (not the end,
        # where base64url padding bits can absorb the change)
        parts = token.split(".")
        sig = list(parts[2])
        mid = len(sig) // 2
        sig[mid] = "X" if sig[mid] != "X" else "Y"
        tampered = parts[0] + "." + parts[1] + "." + "".join(sig)
        assert verify_token(tampered, secret) is None

    def test_wrong_secret_returns_none(self):
        token = create_token("alice", "admin", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
        assert verify_token(token, "wrong-secret-key-that-is-at-least-thirty-two-bytes-long") is None

    def test_garbage_token_returns_none(self):
        assert verify_token("not-a-jwt", "test-secret-key-that-is-at-least-thirty-two-bytes-long") is None

    def test_custom_expiry(self):
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret, expires_hours=1)
        claims = verify_token(token, secret)
        assert claims is not None
        # exp should be roughly 1 hour from now
        exp = datetime.fromtimestamp(claims["exp"], tz=UTC)
        now = datetime.now(UTC)
        delta = exp - now
        assert timedelta(minutes=55) < delta < timedelta(minutes=65)

    def test_programming_errors_propagate(self):
        """Only jwt validation errors map to None; bugs must raise.

        Passing None as the secret is a programming error (TypeError from
        PyJWT), not an invalid token -- it must not be swallowed into a 401.
        """
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret)
        with pytest.raises(TypeError):
            verify_token(token, None)

    def test_default_expiry_is_720_hours(self):
        secret = "test-secret-key-that-is-at-least-thirty-two-bytes-long"
        token = create_token("alice", "admin", secret)
        claims = verify_token(token, secret)
        assert claims is not None
        exp = datetime.fromtimestamp(claims["exp"], tz=UTC)
        iat = datetime.fromtimestamp(claims["iat"], tz=UTC)
        delta = exp - iat
        assert delta == timedelta(hours=720)

python

class TestPasswordHashing:
    """hash_password and verify_password."""

    def test_hash_and_verify_correct(self):
        hashed = hash_password("my-password")
        assert verify_password("my-password", hashed) is True

    def test_verify_wrong_password(self):
        hashed = hash_password("correct-password")
        assert verify_password("wrong-password", hashed) is False

    def test_different_hashes_for_same_password(self):
        h1 = hash_password("same")
        h2 = hash_password("same")
        # bcrypt salts differ, so hashes differ
        assert h1 != h2
        # But both verify
        assert verify_password("same", h1) is True
        assert verify_password("same", h2) is True

    def test_hash_returns_string(self):
        hashed = hash_password("test")
        assert isinstance(hashed, str)
        assert hashed.startswith("$2")

    def test_password_over_72_bytes_rejected(self):
        with pytest.raises(ValueError, match="72-byte limit"):
            hash_password("x" * 73)

    def test_password_exactly_72_bytes_accepted(self):
        pw = "x" * 72
        hashed = hash_password(pw)
        assert verify_password(pw, hashed) is True

    def test_multibyte_password_over_72_bytes_rejected(self):
        # 25 chars but 75 UTF-8 bytes -- the limit is byte-based.
        pw = "€" * 25  # euro sign, 3 bytes each
        assert len(pw) < 72
        with pytest.raises(ValueError, match="72-byte limit"):
            hash_password(pw)

python

class TestGetCurrentUser:
    """get_current_user extracts JWT from 3 sources."""

    @pytest.mark.anyio
    async def test_bearer_header(self):
        async def handler(request, user=None):
            return JSONResponse({"sub": user["sub"], "role": user["role"]})

        app = _make_auth_app(handler, deps={"user": get_current_user})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("alice", "admin", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(
                    "/protected",
                    headers={"Authorization": f"Bearer {token}"},
                )
            assert resp.status_code == 200
            assert resp.json()["sub"] == "alice"
            assert resp.json()["role"] == "admin"
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_session_cookie(self):
        async def handler(request, user=None):
            return JSONResponse({"sub": user["sub"]})

        app = _make_auth_app(handler, deps={"user": get_current_user})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("bob", "user", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(
                    "/protected",
                    cookies={"session": token},
                )
            assert resp.status_code == 200
            assert resp.json()["sub"] == "bob"
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_query_param_rejected_by_default(self):
        """?token= leaks tokens into logs/history -- off unless opted in."""
        async def handler(request, user=None):
            return JSONResponse({"sub": user["sub"]})

        app = _make_auth_app(handler, deps={"user": get_current_user})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("carol", "viewer", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(f"/protected?token={token}")
            assert resp.status_code == 401
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_query_param_opt_in(self):
        async def handler(request, user=None):
            return JSONResponse({"sub": user["sub"]})

        def dep(request):
            return get_current_user(request, allow_query_token=True)

        app = _make_auth_app(handler, deps={"user": dep})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("carol", "viewer", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(f"/protected?token={token}")
            assert resp.status_code == 200
            assert resp.json()["sub"] == "carol"
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_no_token_returns_401(self):
        async def handler(request, user=None):
            return JSONResponse({"ok": True})

        app = _make_auth_app(handler, deps={"user": get_current_user})
        task, shutdown = await _start_lifespan(app)
        try:
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get("/protected")
            assert resp.status_code == 401
            assert resp.json()["detail"] == "Not authenticated"
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_invalid_token_returns_401(self):
        async def handler(request, user=None):
            return JSONResponse({"ok": True})

        app = _make_auth_app(handler, deps={"user": get_current_user})
        task, shutdown = await _start_lifespan(app)
        try:
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(
                    "/protected",
                    headers={"Authorization": "Bearer invalid-token"},
                )
            assert resp.status_code == 401
            assert resp.json()["detail"] == "Invalid or expired token"
        finally:
            shutdown.set()
            await task

python

class TestRequireRole:
    """require_role(role) factory checks claims["role"]."""

    @pytest.mark.anyio
    async def test_correct_role_passes(self):
        async def handler(request, user=None):
            return JSONResponse({"sub": user["sub"]})

        app = _make_auth_app(handler, deps={"user": require_role("admin")})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("alice", "admin", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(
                    "/protected",
                    headers={"Authorization": f"Bearer {token}"},
                )
            assert resp.status_code == 200
            assert resp.json()["sub"] == "alice"
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_wrong_role_returns_403(self):
        async def handler(request, user=None):
            return JSONResponse({"ok": True})

        app = _make_auth_app(handler, deps={"user": require_role("admin")})
        task, shutdown = await _start_lifespan(app)
        try:
            token = create_token("bob", "viewer", "test-secret-key-that-is-at-least-thirty-two-bytes-long")
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get(
                    "/protected",
                    headers={"Authorization": f"Bearer {token}"},
                )
            assert resp.status_code == 403
            assert "admin" in resp.json()["detail"]
        finally:
            shutdown.set()
            await task

    @pytest.mark.anyio
    async def test_no_token_returns_401(self):
        async def handler(request, user=None):
            return JSONResponse({"ok": True})

        app = _make_auth_app(handler, deps={"user": require_role("admin")})
        task, shutdown = await _start_lifespan(app)
        try:
            async with AsyncClient(
                transport=ASGITransport(app=app), base_url="http://test",
            ) as client:
                resp = await client.get("/protected")
            assert resp.status_code == 401
        finally:
            shutdown.set()
            await task

python

class TestCSRFMiddleware:
    """CSRFMiddleware validates double-submit cookie pattern."""

    @pytest.mark.anyio
    async def test_get_always_passes(self, client_for):
        router = Router()

        @router.get("/data")
        async def data(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        async with client_for(app) as client:
            resp = await client.get("/data")
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_post_without_csrf_returns_403(self, client_for):
        router = Router()

        @router.post("/submit")
        async def submit(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        async with client_for(app) as client:
            resp = await client.post("/submit", json={"data": "test"})
        assert resp.status_code == 403
        assert "CSRF" in resp.json()["detail"]

    @pytest.mark.anyio
    async def test_post_with_matching_csrf_passes(self, client_for):
        csrf_val = "my-csrf-token-value"
        router = Router()

        @router.post("/submit")
        async def submit(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        async with client_for(app) as client:
            resp = await client.post(
                "/submit",
                json={"data": "test"},
                headers={"X-CSRF-Token": csrf_val},
                cookies={"csrf_token": csrf_val},
            )
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_post_with_mismatched_csrf_returns_403(self, client_for):
        router = Router()

        @router.post("/submit")
        async def submit(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        async with client_for(app) as client:
            resp = await client.post(
                "/submit",
                json={"data": "test"},
                headers={"X-CSRF-Token": "value-a"},
                cookies={"csrf_token": "value-b"},
            )
        assert resp.status_code == 403
        assert "mismatch" in resp.json()["detail"]

    @pytest.mark.anyio
    async def test_bearer_token_bypasses_csrf(self, client_for):
        router = Router()

        @router.post("/api/data")
        async def api_data(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        # A real JWT has 3 dot-separated segments
        fake_jwt = "header.payload.signature"
        async with client_for(app) as client:
            resp = await client.post(
                "/api/data",
                json={"data": "test"},
                headers={"Authorization": f"Bearer {fake_jwt}"},
            )
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_exempt_paths(self, client_for):
        router = Router()

        @router.post("/auth/login")
        async def login(request):
            return JSONResponse({"ok": True})

        middleware = [lambda app: CSRFMiddleware(app, exempt_paths=["/auth/login"])]
        app = create_app(router, middleware=middleware)
        async with client_for(app) as client:
            resp = await client.post("/auth/login", json={"user": "test"})
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_disabled_flag(self, client_for):
        router = Router()

        @router.post("/submit")
        async def submit(request):
            return JSONResponse({"ok": True})

        middleware = [lambda app: CSRFMiddleware(app, disabled=True)]
        app = create_app(router, middleware=middleware)
        async with client_for(app) as client:
            resp = await client.post("/submit", json={"data": "test"})
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_head_and_options_exempt(self, client_for):
        router = Router()

        @router.get("/data")
        async def data(request):
            return JSONResponse({"ok": True})

        app = create_app(router, middleware=[CSRFMiddleware])
        async with client_for(app) as client:
            resp = await client.head("/data")
            # HEAD may get 404 since only GET is registered, but CSRF
            # should not be the blocker -- the status should not be 403
            assert resp.status_code != 403

            resp = await client.options("/data")
            assert resp.status_code != 403

    @pytest.mark.anyio
    async def test_websocket_passes_through(self):
        """CSRF middleware ignores websocket scope."""
        router = Router()

        @router.ws("/ws/test")
        async def ws_handler(ws):
            await ws.accept()
            await ws.send_text("hello")
            await ws.close()

        app = create_app(router, middleware=[CSRFMiddleware])

        sent = []
        step = 0

        async def fake_receive():
            nonlocal step
            step += 1
            if step == 1:
                return {"type": "websocket.connect"}
            await asyncio.sleep(100)

        async def fake_send(msg):
            sent.append(msg)

        scope = {
            "type": "websocket",
            "path": "/ws/test",
            "headers": [],
            "query_string": b"",
        }
        await app(scope, fake_receive, fake_send)
        assert sent[0]["type"] == "websocket.accept"

python

class TestRateLimiting:
    """@rate_limit decorator with token bucket."""

    @pytest.mark.anyio
    async def test_within_limit_passes(self, client_for):
        router = Router()

        @router.get("/search")
        @rate_limit("5/minute")
        async def search(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            for _ in range(5):
                resp = await client.get("/search")
                assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_exceeding_limit_returns_429(self, client_for):
        router = Router()

        @router.get("/search")
        @rate_limit("3/minute")
        async def search(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            for _ in range(3):
                resp = await client.get("/search")
                assert resp.status_code == 200
            # 4th request should be rate limited
            resp = await client.get("/search")
            assert resp.status_code == 429
            assert "Rate limit" in resp.json()["detail"]

    @pytest.mark.anyio
    async def test_unlimited_endpoint_unaffected(self, client_for):
        router = Router()

        @router.get("/unlimited")
        async def unlimited(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            for _ in range(20):
                resp = await client.get("/unlimited")
                assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_custom_key_func(self, client_for):
        router = Router()

        @router.get("/api")
        @rate_limit("2/minute", key_func=lambda req: "global")
        async def api(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/api")
            assert resp.status_code == 200
            resp = await client.get("/api")
            assert resp.status_code == 200
            # 3rd request with same global key hits limit
            resp = await client.get("/api")
            assert resp.status_code == 429

    def test_invalid_rate_format_raises(self):
        with pytest.raises(ValueError, match="Invalid rate format"):
            rate_limit("invalid")

    def test_valid_rate_formats(self):
        # These should not raise
        rate_limit("10/second")
        rate_limit("100/minute")
        rate_limit("1000/hour")

    @pytest.mark.anyio
    async def test_stale_buckets_evicted(self, monkeypatch):
        """Buckets idle longer than the rate window are evicted, so the
        bucket dict cannot grow without bound (memory leak / churn-DoS)."""
        import fastware.auth as auth_mod

        clock = {"now": 1000.0}
        monkeypatch.setattr(auth_mod.time, "monotonic", lambda: clock["now"])

        class FakeReq:
            def __init__(self, key):
                self.key = key

        @rate_limit("5/second", key_func=lambda r: r.key)
        async def handler(request):
            return "ok"

        for i in range(100):
            await handler(FakeReq(f"client-{i}"))
        assert len(handler._buckets) == 100

        # All 100 clients go idle past the 1-second window.
        clock["now"] += 2.0
        await handler(FakeReq("fresh-client"))
        assert len(handler._buckets) == 1
        assert set(handler._buckets) == {"fresh-client"}

    @pytest.mark.anyio
    async def test_wrapper_preserves_signature_for_dep_filtering(self, client_for):
        """create_app filters resolved deps by handler signature. The
        rate_limit wrapper must not present a (*args, **kwargs) signature,
        or router-level deps get force-fed to handlers that don't accept
        them (TypeError -> 500)."""
        router = Router()

        @router.get("/limited", deps={"extra": lambda: 42})
        @rate_limit("5/minute")
        async def limited(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/limited")
        assert resp.status_code == 200

    @pytest.mark.anyio
    async def test_wrapper_passes_deps_handler_accepts(self, client_for):
        router = Router()

        @router.get("/limited", deps={"extra": lambda: 42})
        @rate_limit("5/minute")
        async def limited(request, extra=None):
            return JSONResponse({"extra": extra})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/limited")
        assert resp.status_code == 200
        assert resp.json()["extra"] == 42

    def test_wrapper_uses_functools_wraps(self):
        async def my_handler(request):
            """My docstring."""

        decorated = rate_limit("5/minute")(my_handler)
        assert decorated.__name__ == "my_handler"
        assert decorated.__doc__ == "My docstring."
        assert decorated.__wrapped__ is my_handler

    @pytest.mark.anyio
    async def test_per_second_rate(self, client_for):
        router = Router()

        @router.get("/fast")
        @rate_limit("2/second")
        async def fast(request):
            return JSONResponse({"ok": True})

        app = create_app(router)
        async with client_for(app) as client:
            resp = await client.get("/fast")
            assert resp.status_code == 200
            resp = await client.get("/fast")
            assert resp.status_code == 200
            # 3rd immediate request should fail
            resp = await client.get("/fast")
            assert resp.status_code == 429

#Practical Example: JWT Auth on a Route

Setting up JWT authentication on a protected route using the dependency injection system. This example shows how to hash and verify passwords with bcrypt, issue JWT tokens at login, read them back with get_current_user, and enforce role-based access control on specific endpoints:

python

from fastware import Router, create_app
from fastware.auth import (
    create_token,
    get_current_user,
    require_role,
    hash_password,
    verify_password,
    CSRFMiddleware,
    set_session_cookies,
    clear_session_cookies,
)

router = Router()


@router.post("/login")
async def login(request):
    """Authenticate a user and return a JWT token with session cookies."""
    data = request.json
    username = data["username"]
    password = data["password"]

    # Look up the user (your storage layer here)
    user = user_store.find_user(username)
    if not user or not verify_password(password, user["password_hash"]):
        raise HTTPError(401, "Invalid credentials")

    # Create a JWT token
    jwt_secret = request.state["config"]["jwt_secret"]
    token = create_token(username, user["role"], jwt_secret)
    csrf_token = secrets.token_urlsafe(32)

    return JSONResponse(
        {"token": token, "username": username},
        cookies=set_session_cookies(token, csrf_token),
    )


@router.get("/profile", deps={"user": get_current_user})
async def profile(request, user):
    """Return the authenticated user's profile."""
    return {"username": user["sub"], "role": user["role"]}


@router.get("/admin", deps={"user": require_role("admin")})
async def admin_dashboard(request, user):
    """Admin-only endpoint using role-based access control."""
    return {"message": f"Welcome, admin {user['sub']}"}


# Apply CSRF protection as middleware
app = create_app(
    router,
    middleware=[
        lambda app: CSRFMiddleware(app, exempt_paths=["/login"]),
    ],
)

The get_current_user dependency reads the JWT from the Authorization header, session cookie, or ?token= query parameter (in that order). It validates the token against the secret stored in request.state["config"]["jwt_secret"] and returns the decoded claims dict. The require_role factory wraps get_current_user with an additional role check.

#Auth API Reference

#src.fastware.auth

#create_token

#verify_token

#hash_password

#verify_password

#UserStore

#load_users

#save_users

#find_user

#create_user

#delete_user

#JSONFileUserStore

#load_users

#save_users

#get_current_user

#require_role

#CSRFMiddleware

#set_session_cookies

#clear_session_cookies

#rate_limit

#Practical Example: JWT Auth on a Route

#`src.fastware.auth`

#`create_token`

#`verify_token`

#`hash_password`

#`verify_password`

#`UserStore`

#`load_users`

#`save_users`

#`find_user`

#`create_user`

#`delete_user`

#`JSONFileUserStore`

#`load_users`

#`save_users`

#`get_current_user`

#`require_role`

#`CSRFMiddleware`

#`set_session_cookies`

#`clear_session_cookies`

#`rate_limit`