fastware v0.6.0 /src.fastware.auth
Updated
On this page

JWT token creation/verification (HS256), bcrypt hashing, user stores, get_current_user/require_role DI, CSRF middleware, and rate limiting.

#src.fastware.auth

#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:

  1. Authorization: Bearer header
  2. session cookie
  3. ?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.

Search