🔐 Authentication

Multi-user tenant authentication with Google OAuth, CSRF protection, and automatic tenant provisioning.

---

⏰ Sliding Session Configuration

Configure session timeout behavior with sliding expiry.

Parameter	Default	Description
max_age	3600	Session expires after this many seconds of inactivity
sliding	True	Refresh session on each request
absolute_max	None	Optional hard limit regardless of activity (e.g., 86400 for 24h)
secure	True	HTTPS-only cookies in production
same_site	"lax"	SameSite cookie policy for CSRF protection

How Sliding Sessions Work

Current (Fixed Timeout):
  Login at 10:00 → Expires at 11:00 (regardless of activity)
  User active at 10:55 → STILL logged out at 11:00 ❌

Sliding Timeout:
  Login at 10:00 → Expires at 11:00
  User active at 10:55 → Expires at 11:55 ✓
  User active at 11:50 → Expires at 12:50 ✓
  User idle for 1 hour → Logged out ✓

---

🔄 Sliding Session Middleware

Custom middleware that extends Starlette’s SessionMiddleware to refresh cookie max_age on each authenticated request.

---

source

SessionConfig

def SessionConfig(
    max_age:int=3600, sliding:bool=True, absolute_max:int=None, secure:bool=True, same_site:str='lax',
    http_only:bool=True
)->None:

Configuration for sliding session behavior.

Attributes: max_age: Session expires after this many seconds of inactivity. Default: 3600 (1 hour) sliding: If True, refresh session expiry on each request. Default: True absolute_max: Optional hard limit in seconds regardless of activity. Default: None secure: If True, cookie only sent over HTTPS. Default: True same_site: SameSite cookie policy (‘lax’, ‘strict’, ‘none’). Default: ‘lax’ http_only: If True, cookie not accessible via JavaScript. Default: True

Example: >>> config = SessionConfig() # 1 hour inactivity timeout, sliding enabled >>> config = SessionConfig(max_age=1800) # 30 min inactivity timeout >>> config = SessionConfig(max_age=3600, absolute_max=86400) # 1h sliding, 24h hard limit

---

source

SlidingSessionMiddleware

def SlidingSessionMiddleware(
    app, secret_key:str, session_config:SessionConfig=None, session_cookie:str='session', path:str='/'
):

Session middleware with sliding expiry.

Extends Starlette’s SessionMiddleware to refresh the session cookie max_age on each request, implementing sliding session expiry.

Sessions expire after max_age seconds of INACTIVITY, not from login time.

Args: app: ASGI application secret_key: Secret key for signing cookies session_config: SessionConfig instance for timeout settings session_cookie: Cookie name. Default: ‘session’ path: Cookie path. Default: ‘/’

Example: >>> from starlette.applications import Starlette >>> app = Starlette() >>> config = SessionConfig(max_age=3600) # 1 hour inactivity >>> app = SlidingSessionMiddleware(app, secret_key=‘…’, session_config=config)

---

source

create_session_middleware

def create_session_middleware(
    secret_key:str, session_config:SessionConfig=None, session_cookie:str='session'
)->SlidingSessionMiddleware:

Factory to create SlidingSessionMiddleware for FastHTML apps.

Args: secret_key: Secret key for signing session cookies (required) session_config: SessionConfig instance. Default: SessionConfig.default() session_cookie: Cookie name. Default: ‘session’

Returns: Configured SlidingSessionMiddleware instance

Example: >>> from fasthtml.common import FastHTML >>> >>> # Create app WITHOUT default session middleware >>> app = FastHTML(sess_cls=None) # Disable default sessions >>> >>> # Add sliding session middleware >>> config = SessionConfig(max_age=3600) # 1 hour inactivity >>> app = create_session_middleware(‘your-secret-key’, config)(app) >>> >>> # Or wrap during app creation (recommended) >>> middleware = create_session_middleware(‘your-secret-key’, config) >>> # Then configure FastHTML to use it

Note: FastHTML’s default SessionMiddleware needs to be disabled first. See integration documentation for patterns.

🎯 Overview

Category	Functions	Purpose
⏰ Sliding Sessions	SessionConfig, SlidingSessionMiddleware, create_session_middleware	Sliding session expiry (inactivity timeout)
🛡️ Beforeware	create_auth_beforeware	Protect routes, auto-setup tenant DB
🔑 OAuth Client	get_google_oauth_client	Initialize Google OAuth
🔒 CSRF	generate_oauth_state, verify_oauth_state	Prevent session hijacking
👤 Users	create_or_get_global_user, get_user_membership, verify_membership	User & membership management
🏗️ Provisioning	provision_new_user	Auto-create tenant for new users
📋 Session	create_user_session, get_current_user, clear_session	Session management
🚦 Routing	auth_redirect, route_user_after_login, require_tenant_access	Authorization & routing
🌐 Handlers	handle_login_request, handle_oauth_callback, handle_logout	Route implementations

---

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    OAuth Authentication Flow                     │
├─────────────────────────────────────────────────────────────────┤
│  1. User clicks "Login with Google"                             │
│  2. Generate CSRF state token → store in session                │
│  3. Redirect to Google OAuth                                    │
│  4. Google authenticates → redirects with code + state          │
│  5. Verify CSRF state matches session                           │
│  6. Exchange code for user info                                 │
│  7. Create/get GlobalUser in host DB                            │
│  8. Check membership OR auto-provision new tenant               │
│  9. Create session → redirect to dashboard                      │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                       Tenant Model                              │
├─────────────────────────────────────────────────────────────────┤
│  Many Users → One Tenant (many-to-one)                          │
│  Each user belongs to exactly ONE tenant                        │
│  Each tenant can have MANY users                                │
│  New users auto-create their own tenant                         │
└─────────────────────────────────────────────────────────────────┘

---

📚 Quick Reference

Security Features

Feature	Protection
CSRF State Token	Prevents session hijacking attacks
Membership Validation	Ensures cross-tenant isolation
Audit Logging	Tracks all authentication events

Token Expiry

Token Type	Expiry	Behavior
Google OAuth	1 hour	User must re-login (refresh tokens: future)
Session (sliding)	1 hour inactivity	Refreshes on each request via SlidingSessionMiddleware

💡 Sliding Sessions: Use SlidingSessionMiddleware or create_session_middleware() to keep users logged in while active. Sessions only expire after configured inactivity period.

from nbdev.showdoc import show_doc

---

🎭 Role-Based Access Control

Control access based on user roles within the tenant.

Role Hierarchy

Role	Level	Automatic For
admin	3	Tenant owners
editor	2	—
viewer	1	—

Key Functions

Function	Purpose
has_min_role	Check if user meets minimum role requirement
require_role	Route decorator for role-based protection
get_user_role	Derive effective role from session + tenant DB

---

source

has_min_role

def has_min_role(
    user:dict, required_role:str
)->bool:

Check if user meets the minimum role requirement.

Args: user: User dict with ‘role’ field (from request.state.user) required_role: Minimum role needed (‘admin’, ‘editor’, ‘viewer’)

Returns: True if user’s role >= required_role in hierarchy

Example: >>> user = {‘role’: ‘editor’} >>> has_min_role(user, ‘viewer’) # True - editor > viewer >>> has_min_role(user, ‘admin’) # False - editor < admin

---

source

get_user_role

def get_user_role(
    session:dict, tenant_db:Database=None
)->str:

Derive effective role from session and tenant database.

Rules: 1. Tenant owner (session[‘tenant_role’] == ‘owner’) → ‘admin’ 2. System admin → ‘admin’ 3. Otherwise → lookup TenantUser.local_role from tenant DB 4. Fallback → None (user must be explicitly assigned a role)

Args: session: User session dict tenant_db: Tenant database connection (optional)

Returns: Effective role string: ‘admin’, ‘editor’, ‘viewer’, or None

---

source

require_role

def require_role(
    min_role:str
):

Decorator to protect routes with minimum role requirement.

Args: min_role: Minimum role required (‘admin’, ‘editor’, ‘viewer’)

Returns: Decorator that checks request.state.user[‘role’] Returns 403 Forbidden if user lacks required role

Example: >>> @app.get(‘/admin/settings’) >>> @require_role(‘admin’) >>> def admin_settings(request): … return “Admin only content”

>>> @app.get('/reports')  
>>> @require_role('viewer')  # All authenticated users
>>> def view_reports(request):
...     return "Reports"

---

⚡ Session Caching

For HTMX-heavy apps with many partial requests, the beforeware can cache auth data in the session to avoid database queries on every request.

Configuration

Parameter	Default	Description
session_cache	False	Enable caching user dict in session
session_cache_ttl	300	Cache TTL in seconds (5 minutes)

How It Works

Request arrives
    │
    ▼
Check session cache
    │
    ├── Cache valid? → Use cached user data (0 DB queries)
    │
    └── Cache miss/expired? → Query DB → Update cache

Cache Invalidation

Event	Action
Logout	Automatic (session cleared)
Role change	Call invalidate_auth_cache(session)
TTL expiry	Automatic refresh on next request

---

source

invalidate_auth_cache

def invalidate_auth_cache(
    session:dict
):

Clear the auth cache from session.

Call this when: - User role or permissions change - User is added/removed from tenant - Admin changes user’s local_role

Args: session: User session dict

Example: >>> # After admin changes user role >>> tenant_user.local_role = ‘editor’ >>> tables[‘tenant_users’].update(tenant_user) >>> invalidate_auth_cache(session) # Force fresh lookup

Usage Example

from fh_saas.utils_auth import invalidate_auth_cache

# Enable caching (recommended for HTMX apps)
app = FastHTML(
    before=create_auth_beforeware(
        session_cache=True,
        session_cache_ttl=300  # 5 minutes
    )
)

# After changing user role, invalidate their cache
@app.post('/admin/users/{user_id}/role')
def update_role(request, user_id: str, new_role: str):
    tables = request.state.tables
    user = tables['tenant_users'].get(user_id)
    user.local_role = new_role
    tables['tenant_users'].update(user)
    
    # If changing own role, invalidate cache
    if user_id == request.state.user['user_id']:
        invalidate_auth_cache(request.session)
    
    return "Role updated"

Usage in Routes

from fh_saas.utils_auth import require_role, has_min_role

# Option 1: Decorator for route-level protection
@app.get('/admin/users')
@require_role('admin')
def manage_users(request):
    return "Admin-only user management"

@app.get('/dashboard')
@require_role('viewer')  # All roles can access
def dashboard(request):
    return "Dashboard for all users"

# Option 2: Inline check for conditional logic
@app.get('/data')
def view_data(request):
    user = request.state.user
    
    if has_min_role(user, 'admin'):
        return "All data with admin controls"
    elif has_min_role(user, 'editor'):
        return "Data with edit buttons"
    else:
        return "Read-only data view"

Role Derivation Flow

┌─────────────────────────────────────────────────────────────────┐
│                    Role Derivation (in beforeware)              │
├─────────────────────────────────────────────────────────────────┤
│  1. Check session['tenant_role']                                │
│     └── If 'owner' → effective role = 'admin'                  │
│                                                                 │
│  2. Check session['is_sys_admin']                               │
│     └── If True → effective role = 'admin'                     │
│                                                                 │
│  3. Lookup TenantUser.local_role in tenant DB                   │
│     └── Returns assigned role or None                          │
│                                                                 │
│  4. Attach to request.state.user['role']                        │
│     └── Used by require_role() and has_min_role()              │
└─────────────────────────────────────────────────────────────────┘

---

🛡️ Auth Beforeware

Protect routes by checking for authenticated sessions with auto tenant DB setup.

Function	Purpose
create_auth_beforeware	Factory to create route protection middleware

💡 Use case: Pass to FastHTML’s before= parameter for app-wide authentication

---

source

create_auth_beforeware

def create_auth_beforeware(
    redirect_path:str='/login', session_key:str='user_id', skip:list=None, include_defaults:bool=True,
    setup_tenant_db:bool=True, schema_init:Callable=None, session_cache:bool=False, session_cache_ttl:int=300,
    require_subscription:bool=False, subscription_redirect:str=None, grace_period_days:int=3,
    session_config:SessionConfig=None
):

Create Beforeware that checks for authenticated session and sets up request.state.

Args: redirect_path: Where to redirect unauthenticated users session_key: Session key for user ID skip: List of regex patterns to skip auth include_defaults: Include default skip patterns setup_tenant_db: Auto-setup tenant database on request.state schema_init: Optional callback to initialize tables dict. Signature: (tenant_db: Database) -> dict[str, Table] Result stored in request.state.tables session_cache: Enable caching user dict in session to reduce DB queries. Recommended for HTMX-heavy apps. Default: False session_cache_ttl: Cache TTL in seconds. Default: 300 (5 minutes) require_subscription: If True, check for active subscription and return 402 if not found. Default: False subscription_redirect: Optional URL to redirect if subscription required. If None, returns 402 Payment Required response. grace_period_days: Days to allow access after payment failure (default: 3) session_config: Optional SessionConfig for absolute timeout enforcement. Note: Sliding expiry requires SlidingSessionMiddleware. This parameter only enforces absolute_max if configured.

Returns: Beforeware instance for FastHTML apps

Sets on request.state: - user: dict with user_id, email, tenant_id, role, is_owner - tenant_id: str - tenant_db: Database connection - tables: dict of Table objects (if schema_init provided) - subscription: Subscription object (if require_subscription enabled)

Example: >>> # Basic usage >>> beforeware = create_auth_beforeware()

>>> # With session caching for HTMX apps
>>> beforeware = create_auth_beforeware(
...     session_cache=True,
...     session_cache_ttl=300
... )

>>> # With schema initialization
>>> def get_app_tables(db):
...     return {'users': db.create(User, pk='id')}
>>> beforeware = create_auth_beforeware(schema_init=get_app_tables)

>>> # With subscription requirement
>>> beforeware = create_auth_beforeware(
...     require_subscription=True,
...     subscription_redirect='/pricing'
... )

---

🔑 OAuth Client

Function	Purpose
get_google_oauth_client	Initialize Google OAuth client from env vars

⚠️ Required env vars: GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET

---

source

get_google_oauth_client

def get_google_oauth_client(
    
):

Initialize Google OAuth client with credentials from environment.

---

🔒 CSRF Protection

Prevent session hijacking via state token validation.

Function	Purpose
generate_oauth_state	Create random UUID for CSRF protection
verify_oauth_state	Validate callback state matches session

Attack Without CSRF Protection

1. Attacker initiates OAuth → gets auth code
2. Attacker sends victim: yourapp.com/auth/callback?code=ATTACKER_CODE
3. Victim clicks → session created for attacker's account
4. Victim enters data → attacker sees it all

🛡️ Solution: State token generated at login, verified at callback

---

source

verify_oauth_state

def verify_oauth_state(
    session:dict, callback_state:str
):

Verify OAuth callback state matches stored session state (CSRF protection).

---

source

generate_oauth_state

def generate_oauth_state(
    
):

Generate cryptographically secure random state token for CSRF protection.

---

👤 User Management

Function	Purpose
create_or_get_global_user	Create or retrieve user from host DB
get_user_membership	Get user’s active tenant membership
verify_membership	Validate user has access to tenant

---

source

verify_membership

def verify_membership(
    host_db:HostDatabase, user_id:str, tenant_id:str
)->bool:

Verify user has active membership for specific tenant.

---

source

get_user_membership

def get_user_membership(
    host_db:HostDatabase, user_id:str
):

Get single active membership for user.

---

source

create_or_get_global_user

def create_or_get_global_user(
    host_db:HostDatabase, oauth_id:str, email:str, oauth_info:dict=None
):

Create or retrieve GlobalUser from host database.

---

🏗️ Auto-Provisioning

Create tenant infrastructure for first-time users.

Function	Purpose
provision_new_user	Create tenant DB, catalog entry, membership, and TenantUser

Provisioning Steps

1. Create physical tenant database (PostgreSQL/SQLite)
2. Register tenant in host catalog
3. Create membership (user → tenant, role='owner')
4. Create TenantUser profile (local_role='admin')
5. Initialize core tenant schema
6. Log audit event

💡 Future: Insert payment screen before step 1

---

source

provision_new_user

def provision_new_user(
    host_db:HostDatabase, global_user:GlobalUser
)->str:

Auto-provision new tenant for first-time user.

---

📋 Session Management

Function	Purpose
create_user_session	Populate session after successful OAuth
get_current_user	Extract user info from session
clear_session	Clear all session data (logout)

---

source

clear_session

def clear_session(
    session:dict
):

Clear all session data (logout).

---

source

get_current_user

def get_current_user(
    session:dict
)->dict | None:

Extract current user info from session.

Returns: dict with keys: user_id, email, tenant_id, tenant_role, is_sys_admin

Note: The ‘role’ and ‘is_owner’ fields are added by create_auth_beforeware after deriving the effective role from TenantUser.local_role. Access via request.state.user[‘role’] in routes.

---

source

create_user_session

def create_user_session(
    session:dict, global_user:GlobalUser, membership:Membership
):

Create authenticated session after successful OAuth login.

Sets session keys for user identity and tracking: - user_id, email, tenant_id, tenant_role, is_sys_admin: Identity - login_at: Timestamp when user logged in - session_started_at: Unix timestamp for absolute session timeout tracking

---

🚦 Route Helpers

Function	Purpose
auth_redirect	HTMX-aware redirect to login page
route_user_after_login	Determine redirect URL based on user type
require_tenant_access	Get tenant DB with membership validation

---

source

auth_redirect

def auth_redirect(
    request, redirect_url:str='/login'
):

HTMX-aware redirect for authentication flows.

When HTMX makes a partial request and receives a standard redirect (302/303), it follows the redirect and swaps the response into the target element. This causes the login page to appear inside the partial content area.

This function detects HTMX requests and uses the HX-Redirect header to trigger a full page navigation instead.

Args: request: Starlette request object redirect_url: URL to redirect to (default: ‘/login’)

Returns: Response with appropriate redirect mechanism

Example: python @app.get('/dashboard') def dashboard(request): if not get_current_user(request.session): return auth_redirect(request) return render_dashboard()

---

source

require_tenant_access

def require_tenant_access(
    request_or_session
):

Get tenant database with membership validation.

---

source

route_user_after_login

def route_user_after_login(
    global_user:GlobalUser, membership:Membership=None
)->str:

Determine redirect URL based on user type and membership.

🌐 OAuth Route Handlers

Complete OAuth 2.0 flow handlers:

Function	Purpose
handle_login_request	Initiate OAuth with CSRF protection
handle_oauth_callback	Process provider response
handle_logout	Clear session and redirect

---

source

handle_logout

def handle_logout(
    session
):

Clear session and redirect to login page.

---

source

handle_oauth_callback

def handle_oauth_callback(
    code:str, state:str, request, session
):

Complete OAuth flow: CSRF verify → user info → provision → session → redirect.

---

source

handle_login_request

def handle_login_request(
    request, session
):

Generate Google OAuth URL with CSRF state protection.