
Betterauth Fastapi Jwt Bridge
- 32 installs
- 1 repo stars
- Updated January 27, 2026
- bilalmk/todo_correct
Better Auth + FastAPI JWT Bridge is a skill that implements JWKS-verified JWT authentication and user isolation between a Better Auth Next.js frontend and a FastAPI backend.
About
Better Auth + FastAPI JWT Bridge is a skill that implements stateless JWT authentication between a Better Auth Next.js frontend and a FastAPI backend using JWKS verification. It enables JWT in Better Auth, verifies tokens on the backend, enforces user isolation, and protects API routes so a user can only access their own data. A developer uses it when connecting a Better Auth frontend to a Python backend, including a hybrid String/UUID id architecture and troubleshooting common auth errors.
- Bridges Better Auth (Next.js) and FastAPI with JWKS JWT verification
- Implements user isolation and per-user authorization on API routes
- Provides copy-in templates for jwt_verification.py and auth dependencies
Betterauth Fastapi Jwt Bridge by the numbers
- 32 all-time installs (skills.sh)
- Ranked #1,480 of 2,203 Security skills by installs in the Skillselion catalog
- Data as of Jul 28, 2026 (Skillselion catalog sync)
betterauth-fastapi-jwt-bridge capabilities & compatibility
free (open-source libraries)
- Capabilities
- jwt verification · auth bridge · user isolation · api route protection
- Use cases
- security audit · api development
- Pricing
- Free
What betterauth-fastapi-jwt-bridge says it does
Implement production-ready JWT authentication between Better Auth (Next.js) and FastAPI using JWKS verification for secure, stateless authentication.
Better Auth does NOT provide a `useSession()` hook. Use `authClient.getSession()` with `useEffect`:
pip install fastapi python-jose[cryptography] pyjwt cryptography httpx
npx skills add https://github.com/bilalmk/todo_correct --skill betterauth-fastapi-jwt-bridgeAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 32 |
|---|---|
| repo stars | ★ 1 |
| Last updated | January 27, 2026 |
| Repository | bilalmk/todo_correct ↗ |
What it does
Wire Better Auth (Next.js) JWTs to a FastAPI backend with JWKS verification, user isolation, and protected routes.
Who is it for?
Verifying Better Auth JWTs in a FastAPI backend with user isolation
Skip if: Simple session-only auth without JWT/JWKS or a Python backend
When should I use this skill?
You need to integrate Better Auth with a FastAPI backend using JWT/JWKS verification
What you get
FastAPI routes that verify Better Auth JWTs via JWKS and enforce per-user authorization.
- jwt_verification.py and auth dependencies for FastAPI
- Frontend api-client.ts
- JWKS-protected API routes
By the numbers
- Two separate required tables: session (token column) and jwks
- 5-step quick-start workflow
Files
Better Auth + FastAPI JWT Bridge
Implement production-ready JWT authentication between Better Auth (Next.js) and FastAPI using JWKS verification for secure, stateless authentication.
Architecture
User Login (Frontend)
↓
Better Auth → Issues JWT Token
↓
Frontend API Request → Authorization: Bearer <token>
↓
FastAPI Backend → Verifies JWT with JWKS → Returns filtered dataQuick Start Workflow
Step 1: Enable JWT in Better Auth (Frontend)
// lib/auth.ts
import { betterAuth } from "better-auth"
import { jwt } from "better-auth/plugins"
export const auth = betterAuth({
plugins: [jwt()], // Enables JWT + JWKS endpoint
// ... other config
})Database Migration Required:
After adding JWT plugin, run migrations to create required tables:
# Next.js (Better Auth CLI)
npx @better-auth/cli migrate⚠️ IMPORTANT - Two Separate Tables Required:
1. `session` table must have `token` column (core Better Auth requirement)
- Error:
column "token" of relation "session" does not exist - Fix: See Database Schema Issues
2. `jwks` table must exist (JWT plugin requirement)
- Error:
relation "jwks" does not exist - Fix: See Database Schema Issues
These are separate migrations. The JWT plugin creates the jwks table but does NOT modify the session table.
Step 2: Verify JWKS Endpoint
Test the JWKS endpoint is working:
python scripts/verify_jwks.py http://localhost:3000/api/auth/jwksStep 3: Implement Backend Verification
Copy templates from assets/ to your FastAPI project:
assets/jwt_verification.py→backend/app/auth/jwt_verification.pyassets/auth_dependencies.py→backend/app/auth/dependencies.py
Install dependencies:
pip install fastapi python-jose[cryptography] pyjwt cryptography httpxStep 4: Protect API Routes
from app.auth.dependencies import verify_user_access
@router.get("/{user_id}/tasks")
async def get_tasks(
user_id: str,
user: dict = Depends(verify_user_access)
):
# user_id is verified to match authenticated user
return get_user_tasks(user_id)Step 5: Configure Frontend API Client
Copy assets/api_client.ts to frontend/lib/api-client.ts and use:
import { getTasks, createTask } from "@/lib/api-client"
const tasks = await getTasks(userId)⚠️ React Component Pattern:
Better Auth does NOT provide a useSession() hook. Use authClient.getSession() with useEffect:
import { useState, useEffect } from "react"
import { authClient } from "@/lib/auth-client"
function MyComponent() {
const [user, setUser] = useState(null)
useEffect(() => {
async function loadSession() {
const session = await authClient.getSession()
if (session?.data?.user) {
setUser(session.data.user)
}
}
loadSession()
}, [])
return <div>Welcome {user?.name}</div>
}See Frontend Integration Issues for complete examples.
Better Auth UUID Integration (Hybrid ID Architecture)
Problem Solved: Better Auth uses String IDs internally, but applications often need UUID for type consistency across API routes and database foreign keys.
Solution: Hybrid ID approach - User table has both id (String, Better Auth requirement) and uuid (UUID, application use).
Database Schema
CREATE TABLE "user" (
id VARCHAR PRIMARY KEY, -- Better Auth String ID
uuid UUID UNIQUE NOT NULL, -- Application UUID ⭐
email VARCHAR UNIQUE NOT NULL,
"emailVerified" BOOLEAN DEFAULT FALSE,
name VARCHAR,
"createdAt" TIMESTAMP NOT NULL,
"updatedAt" TIMESTAMP NOT NULL
);
-- UUID auto-generated by database
ALTER TABLE "user" ALTER COLUMN uuid SET DEFAULT gen_random_uuid();
-- All foreign keys point to user.uuid
CREATE TABLE tasks (
id UUID PRIMARY KEY,
user_id UUID REFERENCES "user"(uuid) ON DELETE CASCADE, -- ⭐ FK to uuid
title VARCHAR NOT NULL,
...
);Frontend Configuration (Better Auth)
Add UUID generation hook and JWT custom claim:
// lib/auth.ts
import { betterAuth } from "better-auth"
import { jwt } from "better-auth/plugins"
import { Pool } from "pg"
const pool = new Pool({ connectionString: process.env.DATABASE_URL })
export const auth = betterAuth({
database: pool,
// Hook to fetch database-generated UUID
hooks: {
user: {
created: async ({ user }) => {
const result = await pool.query(
'SELECT uuid FROM "user" WHERE id = $1',
[user.id]
)
const uuid = result.rows[0]?.uuid
return { ...user, uuid }
}
}
},
// Include UUID in JWT payload
plugins: [
jwt({
algorithm: "EdDSA",
async jwt(user, session) {
return {
uuid: user.uuid, // ⭐ Custom claim for backend
}
},
}),
],
})Backend Pattern (FastAPI)
Extract UUID from JWT custom claim (not sub):
# backend/app/auth/dependencies.py
from uuid import UUID
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
payload = verify_jwt_token(token)
# Extract UUID from custom claim (not 'sub')
user_uuid_str = payload.get("uuid") # ⭐
user_uuid = UUID(user_uuid_str)
# Query by UUID
user = await session.execute(
select(User).where(User.uuid == user_uuid)
)
return user.scalar_one_or_none()
async def verify_user_match(
user_id: UUID, # From URL path
current_user: User = Depends(get_current_user)
) -> User:
# Compare UUIDs (not String IDs)
if current_user.uuid != user_id:
raise HTTPException(403, "Not authorized")
return current_userKey Pattern: Always query by User.uuid and validate against UUID from JWT custom claim.
Key Components
1. JWKS Verification Flow
1. Fetch JWKS (cached) from Better Auth endpoint 2. Extract kid (key ID) from JWT token header 3. Find matching public key in JWKS by kid 4. Verify signature using Ed25519 public key 5. Validate claims (issuer, audience, expiration) 6. Extract user info from payload (sub claim)
2. User Isolation Pattern
Always verify user_id from JWT matches user_id in URL:
if current_user["user_id"] != user_id:
raise HTTPException(status_code=403, detail="Not authorized")This prevents users from accessing other users' data.
3. JWT Payload Structure (Updated with UUID Integration)
{
"sub": "user_abc123", // Better Auth String ID
"uuid": "a1b2c3d4-e5f6...", // Application UUID (custom claim) ⭐
"email": "user@example.com",
"name": "User Name",
"iat": 1234567890, // Issued at
"exp": 1234567890, // Expiration
"iss": "http://localhost:3000",
"aud": "http://localhost:3000"
}Important: The uuid custom claim is used for backend user identification and database queries. Better Auth manages users with String IDs (sub), while the application uses UUIDs (uuid) for type consistency.
Environment Configuration
Frontend (.env.local):
BETTER_AUTH_SECRET="min-32-chars-secret"
BETTER_AUTH_URL="http://localhost:3000"
NEXT_PUBLIC_API_URL="http://localhost:8000"Backend (.env):
BETTER_AUTH_URL="http://localhost:3000"
DATABASE_URL="postgresql://..."Testing & Validation
Test JWKS Endpoint
python scripts/verify_jwks.py http://localhost:3000/api/auth/jwksExpected output shows public keys with kid, kty, crv, and x fields.
Test JWT Verification
python scripts/test_jwt_verification.py \
--jwks-url http://localhost:3000/api/auth/jwks \
--token "eyJhbGci..."Troubleshooting
Authentication Issues
| Issue | Solution |
|---|---|
| "relation 'jwks' does not exist" | Create JWKS table migration - see Database Schema Issues |
| "column 'token' does not exist" | Add token column to session table - see Database Schema Issues |
| "Token missing UUID (uuid claim)" | Configure Better Auth hook and JWT plugin - see UUID Integration Issues |
| "User not found after registration" | Dual auth system conflict - see UUID Integration Issues |
| "authClient.useSession is not a function" | Use authClient.getSession() in useEffect - see Frontend Integration Issues |
| "No authentication token available" | Use session.data.session.token not session.session.token - see Frontend Integration Issues |
| "Unable to find matching signing key" | Clear JWKS cache in jwt_verification.py |
| "Token has expired" | Frontend needs to refresh session |
| "Invalid token claims" | Check issuer/audience match BETTER_AUTH_URL |
| 403 Forbidden (UUID mismatch) | Ensure UUID comparison, not String vs UUID - see UUID Integration Issues |
Frontend-Backend Integration Issues (NEW - 2026-01-02)
| Issue | Root Cause | Solution |
|---|---|---|
| Tasks not displaying despite 200 OK | Backend returns array, frontend expects paginated object | Handle both formats with Array.isArray() check - see Frontend-Backend Integration |
| Tag filtering crashes | Backend returns tag objects {id, name, color}, frontend expected number[] | Update TypeScript types to match Pydantic schemas - see Tag Filtering |
| Pagination shows "NaN" | Optional priority field used in arithmetic without null check | Add null checks with defaults for optional fields - see Priority Sorting |
| Tags not saving to database | TaskCreate schema doesn't accept tags field | Use multi-step operation: create task, then assign tags - see Tag Assignment |
| Edit form fields blank | Uncontrolled components + field name mismatches + datetime format | Use controlled components, match field names, convert datetime - see Edit Form |
| 500 Error: timezone comparison | Comparing offset-naive and offset-aware datetimes | Normalize both to UTC before comparison - see Timezone Fix |
| Tag color validation fails | Frontend required color, backend allows optional | Make color optional in Zod schema, provide defaults - see Tag Color |
| Tag filter checkboxes broken | Backend returns id: number, FilterContext uses string[] | Convert IDs to strings for comparison - see Tag Filters |
📚 Critical Reading: See Frontend-Backend Integration Issues section in troubleshooting guide for detailed fixes with code examples. This section documents 8 critical issues discovered during implementation and their resolutions.
Key Learnings: 1. Always read backend Pydantic schemas before writing frontend types 2. Handle optional fields with null checks and defaults 3. Use controlled components for pre-filled forms 4. Match field names exactly between frontend and backend 5. Test with actual backend responses, not mocked data
See references/troubleshooting.md for detailed solutions and prevention strategies.
Advanced Topics
JWKS Caching Strategy
The implementation uses @lru_cache to cache JWKS responses:
- Cache invalidated if token has unknown
kid - Public keys rarely change (safe to cache)
- Reduces network calls to Better Auth
See references/jwks-approach.md for implementation details.
Security Checklist
Before production:
- ✅ HTTPS only for all API calls
- ✅ Token expiration validated
- ✅ Issuer/audience claims verified
- ✅ User ID authorization enforced
- ✅ CORS properly configured
- ✅ Error messages don't leak sensitive info
See references/security-checklist.md for complete list.
Resources
scripts/
verify_jwks.py- Test JWKS endpoint availabilitytest_jwt_verification.py- Validate JWT token verification
references/
jwks-approach.md- Detailed JWKS implementation guidesecurity-checklist.md- Production security requirementstroubleshooting.md- Common issues and fixes
assets/
jwt_verification.py- Complete JWKS verification module templateauth_dependencies.py- FastAPI dependencies templateapi_client.ts- Frontend API client templatebetter_auth_migrations.py- Alembic migration templates for Better Auth tables (including token column fix)
Why JWKS Over Shared Secret?
| Aspect | JWKS | Shared Secret |
|---|---|---|
| Security | ✅ Asymmetric (more secure) | ⚠️ Symmetric (less secure) |
| Scalability | ✅ Multiple backends | ⚠️ Secret must be shared |
| Production | ✅ Recommended | ⚠️ Development only |
| Complexity | Medium | Simple |
Recommendation: Always use JWKS for production.
/**
* API Client for authenticated requests to FastAPI backend.
*
* This module provides utilities for making authenticated API requests
* from Next.js frontend to FastAPI backend using Better Auth JWT tokens.
*
* Usage:
* Copy this file to: frontend/lib/api-client.ts
*
* Example:
* import { getTasks, createTask } from "@/lib/api-client"
*
* const tasks = await getTasks(userId)
* const newTask = await createTask(userId, { title: "New task" })
*/
import { authClient } from "@/lib/auth-client"
// API base URL from environment variables
const API_BASE_URL = process.env.NEXT_PUBLIC_API_URL || "http://localhost:8000"
/**
* Error thrown when API request fails
*/
export class APIError extends Error {
constructor(
public status: number,
public statusText: string,
message: string
) {
super(message)
this.name = "APIError"
}
}
/**
* Get JWT token from current Better Auth session.
*
* @returns JWT token string or null if not authenticated
*/
async function getAuthToken(): Promise<string | null> {
const session = await authClient.getSession()
// Better Auth client returns session in session.data.session
if (!session?.data?.session) {
return null
}
// Better Auth JWT plugin provides token in session
return session.data.session.token
}
/**
* Make authenticated API request to FastAPI backend.
*
* This function:
* 1. Gets JWT token from Better Auth session
* 2. Adds Authorization header with Bearer token
* 3. Makes the API request
* 4. Handles errors appropriately
*
* @param endpoint - API endpoint path (e.g., "/api/v1/user123/tasks")
* @param options - Fetch options (method, body, headers, etc.)
* @returns Response data as JSON
* @throws APIError if request fails
* @throws Error if not authenticated
*/
export async function apiRequest<T>(
endpoint: string,
options: RequestInit = {}
): Promise<T> {
// Get authentication token
const token = await getAuthToken()
if (!token) {
throw new Error("No authentication token available. Please log in.")
}
// Build full URL
const url = `${API_BASE_URL}${endpoint}`
// Make request with Authorization header
const response = await fetch(url, {
...options,
headers: {
"Authorization": `Bearer ${token}`,
"Content-Type": "application/json",
...options.headers,
},
})
// Handle errors
if (!response.ok) {
// Try to get error details from response
let errorMessage = response.statusText
try {
const errorData = await response.json()
errorMessage = errorData.detail || errorMessage
} catch {
// If response is not JSON, use statusText
}
throw new APIError(response.status, response.statusText, errorMessage)
}
// Return parsed JSON
return response.json()
}
// ============================================================================
// Task API Functions
// ============================================================================
export interface Task {
id: number
user_id: string
title: string
description: string | null
completed: boolean
created_at: string
updated_at: string
}
export interface TaskCreate {
title: string
description?: string
}
export interface TaskUpdate {
title?: string
description?: string
completed?: boolean
}
/**
* Get all tasks for a user.
*
* @param userId - User ID (must match authenticated user)
* @returns Array of tasks
*/
export async function getTasks(userId: string): Promise<Task[]> {
return apiRequest<Task[]>(`/api/v1/${userId}/tasks`)
}
/**
* Get a specific task by ID.
*
* @param userId - User ID (must match authenticated user)
* @param taskId - Task ID
* @returns Task object
*/
export async function getTask(userId: string, taskId: number): Promise<Task> {
return apiRequest<Task>(`/api/v1/${userId}/tasks/${taskId}`)
}
/**
* Create a new task.
*
* @param userId - User ID (must match authenticated user)
* @param taskData - Task data (title and optional description)
* @returns Created task
*/
export async function createTask(
userId: string,
taskData: TaskCreate
): Promise<Task> {
return apiRequest<Task>(`/api/v1/${userId}/tasks`, {
method: "POST",
body: JSON.stringify(taskData),
})
}
/**
* Update an existing task.
*
* @param userId - User ID (must match authenticated user)
* @param taskId - Task ID
* @param taskData - Task data to update
* @returns Updated task
*/
export async function updateTask(
userId: string,
taskId: number,
taskData: TaskUpdate
): Promise<Task> {
return apiRequest<Task>(`/api/v1/${userId}/tasks/${taskId}`, {
method: "PATCH",
body: JSON.stringify(taskData),
})
}
/**
* Delete a task.
*
* @param userId - User ID (must match authenticated user)
* @param taskId - Task ID
*/
export async function deleteTask(
userId: string,
taskId: number
): Promise<void> {
return apiRequest(`/api/v1/${userId}/tasks/${taskId}`, {
method: "DELETE",
})
}
/**
* Toggle task completion status.
*
* @param userId - User ID (must match authenticated user)
* @param taskId - Task ID
* @param completed - New completion status
* @returns Updated task
*/
export async function toggleTaskCompletion(
userId: string,
taskId: number,
completed: boolean
): Promise<Task> {
return updateTask(userId, taskId, { completed })
}
// ============================================================================
// React Hook Example (Optional)
// ============================================================================
/**
* Example React hook for using API client with error handling.
*
* Usage:
* const { data: tasks, loading, error } = useAPIData(() => getTasks(userId))
*
* Note: You can use libraries like SWR or React Query for more features.
*/
import { useState, useEffect } from "react"
export function useAPIData<T>(
fetcher: () => Promise<T>,
dependencies: any[] = []
) {
const [data, setData] = useState<T | null>(null)
const [loading, setLoading] = useState(true)
const [error, setError] = useState<Error | null>(null)
useEffect(() => {
let cancelled = false
async function fetchData() {
try {
setLoading(true)
setError(null)
const result = await fetcher()
if (!cancelled) {
setData(result)
}
} catch (err) {
if (!cancelled) {
setError(err instanceof Error ? err : new Error(String(err)))
}
} finally {
if (!cancelled) {
setLoading(false)
}
}
}
fetchData()
return () => {
cancelled = true
}
}, dependencies)
return { data, loading, error }
}
"""
FastAPI dependencies for authentication with Better Auth UUID integration.
This module provides FastAPI dependencies for JWT authentication and user authorization.
Updated to support Better Auth hybrid ID approach (String ID + UUID).
Usage:
Copy this file to: backend/app/auth/dependencies.py
Example:
from app.auth.dependencies import get_current_user, verify_user_access
from uuid import UUID
@router.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
return {"user_uuid": user["uuid"]} # Use UUID
@router.get("/{user_id}/tasks")
async def get_tasks(
user_id: UUID, # UUID in path
user: dict = Depends(verify_user_access)
):
# user_id is guaranteed to match authenticated user's UUID
return get_user_tasks(user_id)
"""
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Dict, Any
from uuid import UUID
from .jwt_verification import verify_jwt_token, extract_user_from_payload
# OAuth2 scheme for Swagger UI
# This tells FastAPI to look for the token in the Authorization header
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
description="JWT token from Better Auth"
)
async def get_current_user(token: str = Depends(oauth2_scheme)) -> Dict[str, Any]:
"""
FastAPI dependency to get the current authenticated user.
This dependency:
1. Extracts JWT token from Authorization header
2. Verifies token signature using JWKS
3. Validates token claims (exp, iss, aud)
4. Returns user information from token payload
Usage:
@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
return {"user_id": user["user_id"]}
Args:
token: JWT token from Authorization header (injected by FastAPI)
Returns:
User information dictionary:
{
"user_id": "user_abc123", # Better Auth String ID (from 'sub')
"uuid": "a1b2c3d4-e5f6-...", # Application UUID (from 'uuid' claim) ⭐
"email": "user@example.com",
"name": "User Name",
"payload": {...} # Full JWT payload
}
Raises:
HTTPException 401: If token is invalid, expired, or missing
"""
# Verify token using JWKS
payload = verify_jwt_token(token)
# Extract user information
user = extract_user_from_payload(payload)
return user
async def verify_user_access(
user_id: UUID,
current_user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
"""
Verify that the authenticated user's UUID matches the requested user_id.
This prevents users from accessing other users' data by checking that
the UUID in the URL path matches the UUID from the JWT token.
Updated for Better Auth UUID integration: Compares UUID (not String ID).
Usage:
@app.get("/api/v1/{user_id}/tasks")
async def get_tasks(
user_id: UUID, # UUID from URL path
user: dict = Depends(verify_user_access)
):
# If we reach here, user_id is guaranteed to match authenticated user's UUID
return db.query(Task).filter(Task.user_id == user_id).all()
Args:
user_id: User UUID from URL path parameter
current_user: Current authenticated user from JWT (injected by get_current_user)
Returns:
User information if access is granted
Raises:
HTTPException 403: If user tries to access another user's resources
"""
# Compare UUIDs (not String IDs)
current_user_uuid = UUID(current_user["uuid"])
if current_user_uuid != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to access this user's resources",
)
return current_user
async def get_current_active_user(
current_user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
"""
Get the current active user (with additional active/disabled check).
Extend this function if you have a user status in your database.
Usage:
@app.get("/users/me")
async def read_users_me(user: dict = Depends(get_current_active_user)):
return user
Args:
current_user: Current authenticated user from JWT
Returns:
User information if user is active
Raises:
HTTPException 400: If user is disabled (if you implement this check)
"""
# Example: Check if user is disabled in your database
# from app.database import get_db
# user_in_db = db.query(User).filter(User.id == current_user["user_id"]).first()
# if user_in_db and user_in_db.disabled:
# raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def require_admin(
current_user: Dict[str, Any] = Depends(get_current_user)
) -> Dict[str, Any]:
"""
Require admin role for the current user.
Extend this function if you have role-based access control (RBAC).
Usage:
@app.delete("/admin/users/{user_id}")
async def delete_user(
user_id: str,
admin: dict = Depends(require_admin)
):
# Only admins can reach this endpoint
delete_user_from_db(user_id)
Args:
current_user: Current authenticated user from JWT
Returns:
User information if user is admin
Raises:
HTTPException 403: If user is not an admin
"""
# Example: Check if user has admin role
# This depends on how you implement roles in your system
# Option 1: Role in JWT payload
# user_role = current_user["payload"].get("role")
# if user_role != "admin":
# raise HTTPException(status_code=403, detail="Admin access required")
# Option 2: Role in database
# from app.database import get_db
# user_in_db = db.query(User).filter(User.id == current_user["user_id"]).first()
# if not user_in_db or not user_in_db.is_admin:
# raise HTTPException(status_code=403, detail="Admin access required")
# For now, return user (customize based on your role implementation)
return current_user
"""
Better Auth Database Migration Templates (Alembic)
These migrations ensure Better Auth tables have the correct schema for both
core authentication (session management) and JWT plugin functionality.
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# ==============================================================================
# Migration 1: Create Better Auth Core Tables
# ==============================================================================
# Revision ID: cdb86b478398
# Revises: (your previous migration)
# Create Date: 2026-01-02 02:35:50
def create_better_auth_tables_upgrade() -> None:
"""Create all Better Auth tables with CORRECT schema including token column."""
# Create BetterAuth user table
op.create_table(
'user',
sa.Column('id', sa.String(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('emailVerified', sa.Boolean(), nullable=False, server_default='false'),
sa.Column('name', sa.String(), nullable=True),
sa.Column('image', sa.String(), nullable=True),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('updatedAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)
# Create BetterAuth session table
# ⚠️ IMPORTANT: Must include 'token' column!
op.create_table(
'session',
sa.Column('id', sa.String(), nullable=False),
sa.Column('token', sa.String(), nullable=False), # ✅ REQUIRED for Better Auth
sa.Column('userId', sa.String(), nullable=False),
sa.Column('expiresAt', sa.DateTime(), nullable=False),
sa.Column('ipAddress', sa.String(), nullable=True),
sa.Column('userAgent', sa.String(), nullable=True),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('updatedAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.PrimaryKeyConstraint('id'),
sa.ForeignKeyConstraint(['userId'], ['user.id'], ondelete='CASCADE')
)
op.create_index('idx_session_userId', 'session', ['userId'])
# Create BetterAuth account table (for OAuth providers)
op.create_table(
'account',
sa.Column('id', sa.String(), nullable=False),
sa.Column('userId', sa.String(), nullable=False),
sa.Column('accountId', sa.String(), nullable=False),
sa.Column('providerId', sa.String(), nullable=False),
sa.Column('accessToken', sa.Text(), nullable=True),
sa.Column('refreshToken', sa.Text(), nullable=True),
sa.Column('idToken', sa.Text(), nullable=True),
sa.Column('expiresAt', sa.DateTime(), nullable=True),
sa.Column('password', sa.String(), nullable=True),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('updatedAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.PrimaryKeyConstraint('id'),
sa.ForeignKeyConstraint(['userId'], ['user.id'], ondelete='CASCADE'),
sa.UniqueConstraint('providerId', 'accountId')
)
op.create_index('idx_account_userId', 'account', ['userId'])
# Create BetterAuth verification table (for email verification)
op.create_table(
'verification',
sa.Column('id', sa.String(), nullable=False),
sa.Column('identifier', sa.String(), nullable=False),
sa.Column('value', sa.String(), nullable=False),
sa.Column('expiresAt', sa.DateTime(), nullable=False),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('updatedAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.PrimaryKeyConstraint('id')
)
op.create_index('idx_verification_identifier', 'verification', ['identifier'])
def create_better_auth_tables_downgrade() -> None:
"""Drop Better Auth tables in reverse order (due to foreign key constraints)."""
op.drop_index('idx_verification_identifier', table_name='verification')
op.drop_table('verification')
op.drop_index('idx_account_userId', table_name='account')
op.drop_table('account')
op.drop_index('idx_session_userId', table_name='session')
op.drop_table('session')
op.drop_table('user')
# ==============================================================================
# Migration 2: Add Missing 'token' Column (if needed)
# ==============================================================================
# Revision ID: 87756bd28f56
# Revises: cdb86b478398
# Create Date: 2026-01-02 02:59:30
def add_token_column_upgrade() -> None:
"""
Add 'token' column to session table.
Use this migration if you created the session table WITHOUT the token column
and are now getting errors like:
"ERROR [Better Auth]: column 'token' of relation 'session' does not exist"
"""
# Add token column with temporary default value
op.add_column(
'session',
sa.Column('token', sa.String(), nullable=False, server_default='')
)
# Remove server_default (it's only needed during migration)
op.alter_column('session', 'token', server_default=None)
def add_token_column_downgrade() -> None:
"""Remove token column from session table."""
op.drop_column('session', 'token')
# ==============================================================================
# Migration 3: Create JWKS Table (JWT Plugin)
# ==============================================================================
# This migration is for the JWT plugin only. Run AFTER enabling jwt() plugin.
def create_jwks_table_upgrade() -> None:
"""
Create JWKS table for Better Auth JWT plugin.
Required when using the jwt() plugin in Better Auth config.
Stores public/private key pairs for JWT signature verification.
"""
op.create_table(
'jwks',
sa.Column('id', sa.String(), nullable=False),
sa.Column('publicKey', sa.String(), nullable=False),
sa.Column('privateKey', sa.String(), nullable=False),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('expiresAt', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
def create_jwks_table_downgrade() -> None:
"""Drop JWKS table."""
op.drop_table('jwks')
# ==============================================================================
# Usage Instructions
# ==============================================================================
"""
## How to Use These Migrations
### Option 1: Create All Tables at Once (Recommended)
Generate a single migration that creates all Better Auth tables:
```bash
alembic revision -m "create_better_auth_tables"
```
Copy the `create_better_auth_tables_upgrade()` function to the `upgrade()` function,
and the `create_better_auth_tables_downgrade()` function to the `downgrade()` function.
Then run:
```bash
alembic upgrade head
```
### Option 2: Fix Missing Token Column
If you already created the session table but it's missing the token column:
```bash
alembic revision -m "add_token_column_to_session_table"
```
Copy the `add_token_column_upgrade()` and `add_token_column_downgrade()` functions.
Then run:
```bash
alembic upgrade head
```
### Option 3: Add JWT Plugin Support
After enabling the jwt() plugin in Better Auth:
```bash
alembic revision -m "create_jwks_table_for_jwt_plugin"
```
Copy the `create_jwks_table_upgrade()` and `create_jwks_table_downgrade()` functions.
Then run:
```bash
alembic upgrade head
```
## Verification
After running migrations, verify the schema:
```python
# Check session table has token column
python -c "
import asyncio
from sqlalchemy import text
from app.core.database import engine
async def check():
async with engine.begin() as conn:
result = await conn.execute(text('''
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = 'session'
ORDER BY ordinal_position
'''))
print('Session table columns:')
for row in result:
print(f' {row[0]:<20} {row[1]:<20} nullable={row[2]}')
asyncio.run(check())
"
```
Expected output should include:
- token (character varying, NO)
- userId (character varying, NO)
- expiresAt (timestamp without time zone, NO)
- All other session columns
## Common Issues
### "column 'token' does not exist"
- Run the `add_token_column_upgrade()` migration
- Or recreate the session table with the correct schema
### "relation 'jwks' does not exist"
- Run the `create_jwks_table_upgrade()` migration
- Ensure jwt() plugin is enabled in Better Auth config
### "JWKS response missing 'keys' field"
- JWT plugin not enabled in frontend Better Auth config
- Run: npx @better-auth/cli migrate
"""
"""
JWT verification using Better Auth JWKS endpoint.
This module provides complete JWT token verification for FastAPI applications
integrating with Better Auth.
Usage:
Copy this file to: backend/app/auth/jwt_verification.py
Environment Variables Required:
BETTER_AUTH_URL - Better Auth base URL (e.g., http://localhost:3000)
"""
from fastapi import HTTPException, status
from jose import jwt, JWTError
from functools import lru_cache
from typing import Dict, Any
import httpx
import os
import logging
logger = logging.getLogger(__name__)
# Configuration from environment
BETTER_AUTH_URL = os.getenv("BETTER_AUTH_URL", "http://localhost:3000")
JWKS_URL = f"{BETTER_AUTH_URL}/api/auth/jwks"
JWT_ALGORITHM = "EdDSA" # Better Auth uses Ed25519
JWT_AUDIENCE = BETTER_AUTH_URL
JWT_ISSUER = BETTER_AUTH_URL
@lru_cache(maxsize=1)
def get_jwks() -> Dict[str, Any]:
"""
Fetch JWKS from Better Auth endpoint.
This is cached because:
1. Public keys don't change frequently
2. Reduces network calls
3. Better Auth documentation recommends caching
Cache is invalidated if we encounter a token with unknown kid.
Returns:
JWKS data dictionary
Raises:
HTTPException 503: If JWKS endpoint is unreachable
"""
try:
response = httpx.get(JWKS_URL, timeout=5.0)
response.raise_for_status()
jwks_data = response.json()
logger.info(f"Fetched JWKS from {JWKS_URL}")
return jwks_data
except httpx.HTTPError as e:
logger.error(f"Failed to fetch JWKS from {JWKS_URL}: {e}")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Unable to fetch authentication keys",
)
def get_signing_key(token: str, jwks_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract the public key from JWKS that matches the token's kid.
Args:
token: JWT token string
jwks_data: JWKS response from Better Auth
Returns:
Public key from JWKS
Raises:
HTTPException: If kid not found in JWKS
"""
try:
# Get key ID from token header
unverified_header = jwt.get_unverified_header(token)
kid = unverified_header.get("kid")
if not kid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token missing key ID (kid)",
headers={"WWW-Authenticate": "Bearer"},
)
# Find matching key in JWKS
for key in jwks_data.get("keys", []):
if key.get("kid") == kid:
return key
# Key not found - invalidate cache and retry once
logger.warning(f"Key ID {kid} not found in cached JWKS, refreshing cache")
get_jwks.cache_clear()
jwks_data = get_jwks()
for key in jwks_data.get("keys", []):
if key.get("kid") == kid:
return key
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unable to find matching signing key",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError as e:
logger.error(f"Error extracting key from token: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token format",
headers={"WWW-Authenticate": "Bearer"},
)
def verify_jwt_token(token: str) -> Dict[str, Any]:
"""
Verify JWT token using Better Auth's JWKS endpoint.
This function:
1. Fetches JWKS (cached)
2. Extracts kid from token header
3. Finds matching public key
4. Verifies token signature
5. Validates issuer and audience
6. Returns decoded payload
Args:
token: JWT token string
Returns:
Decoded token payload containing user information
Raises:
HTTPException: If token is invalid, expired, or verification fails
"""
try:
# Get JWKS (cached)
jwks_data = get_jwks()
# Get signing key
signing_key = get_signing_key(token, jwks_data)
# Verify and decode token
payload = jwt.decode(
token,
signing_key,
algorithms=[JWT_ALGORITHM],
audience=JWT_AUDIENCE,
issuer=JWT_ISSUER,
options={
"verify_signature": True,
"verify_exp": True, # Verify expiration
"verify_aud": True, # Verify audience
"verify_iss": True, # Verify issuer
}
)
logger.debug(f"Successfully verified token for user: {payload.get('sub')}")
return payload
except jwt.ExpiredSignatureError:
logger.warning("Token has expired")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
)
except jwt.JWTClaimsError as e:
logger.warning(f"Invalid token claims: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token claims",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError as e:
logger.error(f"JWT verification failed: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
def extract_user_from_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract user information from JWT payload.
JWT payload structure from Better Auth (with UUID integration):
{
"sub": "user_abc123", # Better Auth String ID
"uuid": "a1b2c3d4-e5f6-...", # Application UUID (custom claim) ⭐
"email": "user@example.com",
"name": "User Name",
"iat": 1234567890,
"exp": 1234567890,
"iss": "http://localhost:3000",
"aud": "http://localhost:3000"
}
Args:
payload: Decoded JWT payload
Returns:
User information dictionary with both String ID and UUID
Raises:
HTTPException: If required claims are missing
"""
user_id = payload.get("sub")
user_uuid = payload.get("uuid")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token missing user ID (sub claim)",
headers={"WWW-Authenticate": "Bearer"},
)
if not user_uuid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token missing UUID (uuid claim)",
headers={"WWW-Authenticate": "Bearer"},
)
return {
"user_id": user_id, # Better Auth String ID
"uuid": user_uuid, # Application UUID ⭐
"email": payload.get("email"),
"name": payload.get("name"),
"payload": payload, # Include full payload for advanced use cases
}
JWKS Approach - Detailed Implementation Guide
Complete technical reference for implementing JWKS-based JWT verification between Better Auth and FastAPI.
JWKS (JSON Web Key Set) Overview
JWKS is a standard format (RFC 7517) for publishing public keys used to verify JWT signatures.
Key Advantages:
- Asymmetric Cryptography: Private key for signing (Better Auth), public key for verification (FastAPI)
- No Shared Secrets: Backend doesn't need access to signing keys
- Scalability: Multiple backends can verify tokens independently
- Security: Signing key compromise doesn't affect all services
Better Auth JWT Plugin Configuration
Basic Setup
// lib/auth.ts
import { betterAuth } from "better-auth"
import { jwt } from "better-auth/plugins"
export const auth = betterAuth({
database: {
provider: "postgres",
url: process.env.DATABASE_URL
},
plugins: [
jwt({
// Optional configuration
expiresIn: "7d", // Token expiration (default: 7 days)
})
],
baseURL: process.env.BETTER_AUTH_URL || "http://localhost:3000",
})What the JWT Plugin Adds
1. JWKS Endpoint: /api/auth/jwks - Returns public keys 2. Token Generation: JWTs included in Better Auth sessions 3. Key Rotation Support: Automatic handling of multiple keys 4. Standard Compliance: JWT tokens following RFC 7519
Database Schema Requirements
CRITICAL: Better Auth requires specific database tables with exact schemas.
Core Tables (Always Required)
1. `user` table: Stores user accounts 2. `session` table: Stores active sessions
- MUST include `token` column (stores session token used as cookie value)
- Common error if missing:
column "token" of relation "session" does not exist
3. `account` table: Stores OAuth provider accounts 4. `verification` table: Stores email verification tokens
JWT Plugin Table (Required When Using jwt() Plugin)
5. `jwks` table: Stores public/private key pairs for JWT signing
- Created automatically when running Better Auth migrations
- Contains:
id,publicKey,privateKey,createdAt,expiresAt
Session Table Schema (with Token Column)
CREATE TABLE session (
id VARCHAR NOT NULL PRIMARY KEY,
token VARCHAR NOT NULL, -- ✅ REQUIRED: Session token
"userId" VARCHAR NOT NULL,
"expiresAt" TIMESTAMP NOT NULL,
"ipAddress" VARCHAR,
"userAgent" VARCHAR,
"createdAt" TIMESTAMP NOT NULL DEFAULT NOW(),
"updatedAt" TIMESTAMP NOT NULL DEFAULT NOW(),
FOREIGN KEY ("userId") REFERENCES "user"(id) ON DELETE CASCADE
);Running Migrations
Frontend (Better Auth):
# Generate migration files
npx @better-auth/cli generate
# Run migrations
npx @better-auth/cli migrateBackend (FastAPI/Alembic):
# Create migration manually or use templates from assets/better_auth_migrations.py
alembic revision -m "create_better_auth_tables"
# Apply migration
alembic upgrade head⚠️ Common Pitfall: The token column is a core Better Auth requirement, not specific to the JWT plugin. Many developers miss this when manually creating tables, leading to signup/login failures.
JWKS Endpoint Response Format
{
"keys": [
{
"kty": "OKP", // Key Type: Octet Key Pair
"crv": "Ed25519", // Curve: Edwards-curve Digital Signature
"x": "bDHiLTt7u...", // Public key value (base64url encoded)
"kid": "c5c7995d-..." // Key ID (identifies which key signed JWT)
}
]
}Field Descriptions:
kty: Key type (OKP for Ed25519)crv: Cryptographic curve (Ed25519 for Better Auth)x: The actual public key valuekid: Unique identifier for the key
FastAPI Implementation
JWT Verification Module
# backend/app/auth/jwt_verification.py
import httpx
from functools import lru_cache
from jose import jwt, JWTError
from typing import Dict, Any
from fastapi import HTTPException, status
import os
import logging
logger = logging.getLogger(__name__)
# Configuration
BETTER_AUTH_URL = os.getenv("BETTER_AUTH_URL", "http://localhost:3000")
JWKS_URL = f"{BETTER_AUTH_URL}/api/auth/jwks"
JWT_ALGORITHM = "EdDSA" # Better Auth uses Ed25519
JWT_AUDIENCE = BETTER_AUTH_URL
JWT_ISSUER = BETTER_AUTH_URL
@lru_cache(maxsize=1)
def get_jwks() -> Dict[str, Any]:
"""
Fetch JWKS from Better Auth endpoint with caching.
Cache Strategy:
- Cache size: 1 (only cache latest JWKS)
- Cache invalidation: Manual (on unknown kid)
- Timeout: 5 seconds (fail fast)
Returns:
JWKS data dictionary
Raises:
HTTPException 503: If JWKS endpoint is unreachable
"""
try:
response = httpx.get(JWKS_URL, timeout=5.0)
response.raise_for_status()
jwks_data = response.json()
logger.info(f"Fetched JWKS from {JWKS_URL}")
return jwks_data
except httpx.HTTPError as e:
logger.error(f"Failed to fetch JWKS from {JWKS_URL}: {e}")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="Unable to fetch authentication keys",
)
def get_signing_key(token: str, jwks_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Extract public key from JWKS matching token's kid.
Process:
1. Decode token header (without verification)
2. Extract kid (key ID)
3. Find matching key in JWKS
4. If not found, refresh cache and retry once
Args:
token: JWT token string
jwks_data: JWKS response from Better Auth
Returns:
Public key dictionary from JWKS
Raises:
HTTPException 401: If kid missing or key not found
"""
try:
# Get key ID from token header
unverified_header = jwt.get_unverified_header(token)
kid = unverified_header.get("kid")
if not kid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token missing key ID (kid)",
headers={"WWW-Authenticate": "Bearer"},
)
# Find matching key in JWKS
for key in jwks_data.get("keys", []):
if key.get("kid") == kid:
return key
# Key not found - invalidate cache and retry
logger.warning(f"Key ID {kid} not in cached JWKS, refreshing")
get_jwks.cache_clear()
jwks_data = get_jwks()
for key in jwks_data.get("keys", []):
if key.get("kid") == kid:
return key
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unable to find matching signing key",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError as e:
logger.error(f"Error extracting key from token: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token format",
headers={"WWW-Authenticate": "Bearer"},
)
def verify_jwt_token(token: str) -> Dict[str, Any]:
"""
Verify JWT token using JWKS.
Verification Steps:
1. Fetch JWKS (cached)
2. Find matching public key by kid
3. Verify signature with Ed25519
4. Validate expiration (exp claim)
5. Validate audience (aud claim)
6. Validate issuer (iss claim)
7. Return decoded payload
Args:
token: JWT token string
Returns:
Decoded token payload with user information
Raises:
HTTPException 401: If verification fails
"""
try:
# Get JWKS (cached)
jwks_data = get_jwks()
# Get signing key
signing_key = get_signing_key(token, jwks_data)
# Verify and decode token
payload = jwt.decode(
token,
signing_key,
algorithms=[JWT_ALGORITHM],
audience=JWT_AUDIENCE,
issuer=JWT_ISSUER,
options={
"verify_signature": True,
"verify_exp": True,
"verify_aud": True,
"verify_iss": True,
}
)
logger.debug(f"Token verified for user: {payload.get('sub')} (UUID: {payload.get('uuid')})")
return payload
except jwt.ExpiredSignatureError:
logger.warning("Expired token received")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
)
except jwt.JWTClaimsError as e:
logger.warning(f"Invalid token claims: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token claims",
headers={"WWW-Authenticate": "Bearer"},
)
except JWTError as e:
logger.error(f"JWT verification failed: {e}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)JWKS Caching Strategy
Why Cache?
1. Performance: Avoid network call on every request 2. Reliability: Reduces dependency on Better Auth availability 3. Cost: Fewer requests to Better Auth server 4. Safety: Public keys rarely change
Cache Implementation
@lru_cache(maxsize=1)
def get_jwks() -> Dict[str, Any]:
# ... fetch JWKSConfiguration:
maxsize=1: Only cache the latest JWKS response- Thread-safe:
lru_cacheis thread-safe by default - Manual invalidation: Clear cache when kid not found
Cache Invalidation
if kid not in cached_jwks:
get_jwks.cache_clear() # Invalidate cache
jwks_data = get_jwks() # Fetch fresh JWKSWhen to Invalidate: 1. Token has kid not in cached JWKS (key rotation occurred) 2. After deployment (if using in-memory cache)
Cache Refresh Strategy
Option 1: TTL-based (Recommended for Production)
import time
_jwks_cache = {"data": None, "timestamp": 0}
JWKS_TTL = 3600 # 1 hour
def get_jwks_with_ttl() -> Dict[str, Any]:
now = time.time()
if _jwks_cache["data"] is None or (now - _jwks_cache["timestamp"]) > JWKS_TTL:
_jwks_cache["data"] = fetch_jwks()
_jwks_cache["timestamp"] = now
return _jwks_cache["data"]Option 2: On-demand (Simpler)
Use @lru_cache and invalidate only when kid not found (current implementation).
Token Flow Diagram
┌─────────────┐
│ Browser │
└──────┬──────┘
│ 1. Login
↓
┌──────────────┐
│ Better Auth │
│ (Next.js) │
└──────┬───────┘
│ 2. Generate JWT (Ed25519 private key)
│ - Sign with private key
│ - Include kid in header
│ - Set exp, iss, aud claims
↓
┌──────────────┐
│ Browser │
└──────┬───────┘
│ 3. API Request
│ Authorization: Bearer <JWT>
↓
┌──────────────┐
│ FastAPI │
│ Backend │
└──────┬───────┘
│ 4. Verify Token
│ a. Fetch JWKS (cached)
│ b. Find public key by kid
│ c. Verify signature
│ d. Validate claims
↓
┌──────────────┐
│ Return Data │
│ (filtered) │
└──────────────┘Error Handling
Common Verification Errors
1. Invalid Signature
jwt.JWTError: Signature verification failedCause: Token was not signed by Better Auth or was tampered with Solution: Ensure BETTER_AUTH_URL matches issuer, check token integrity
2. Expired Token
jwt.ExpiredSignatureError: Signature has expiredCause: Token's exp claim is in the past Solution: Frontend should refresh session or redirect to login
3. Invalid Audience/Issuer
jwt.JWTClaimsError: Invalid audience/issuerCause: Token aud/iss doesn't match expected values Solution: Verify BETTER_AUTH_URL configuration
4. Missing Kid
HTTPException: Token missing key ID (kid)Cause: Token header doesn't have kid field Solution: Ensure Better Auth JWT plugin is properly configured
5. Unknown Kid
HTTPException: Unable to find matching signing keyCause: Key rotation occurred, cached JWKS outdated Solution: Cache is automatically refreshed, retry request
Security Considerations
1. Algorithm Whitelist
algorithms=[JWT_ALGORITHM] # Only allow EdDSAWhy: Prevents algorithm confusion attacks (e.g., downgrade to HS256)
2. Claim Validation
options={
"verify_signature": True, # Must verify signature
"verify_exp": True, # Must check expiration
"verify_aud": True, # Must check audience
"verify_iss": True, # Must check issuer
}Why: Prevents token reuse, replay attacks, and forgery
3. HTTPS Only
# In production
BETTER_AUTH_URL = "https://your-domain.com" # Must use HTTPSWhy: Prevents token interception
4. Error Message Safety
detail="Could not validate credentials" # Generic message
# NOT: detail=f"Invalid signature: {e}" # Leaks infoWhy: Prevents information disclosure attacks
Testing Checklist
- [ ] JWKS endpoint accessible
- [ ] Public keys have required fields (kid, kty, crv, x)
- [ ] Token verification succeeds with valid token
- [ ] Expired tokens rejected
- [ ] Tampered tokens rejected
- [ ] Wrong audience/issuer rejected
- [ ] Cache invalidation works on key rotation
- [ ] Logging captures auth failures
- [ ] HTTPS enforced in production
- [ ] Error messages don't leak sensitive data
Security Checklist - Production Deployment
Comprehensive security checklist for Better Auth + FastAPI JWT integration in production environments.
Pre-Deployment Checklist
🔐 1. Transport Security
- [ ] HTTPS Only: All API endpoints use HTTPS (not HTTP)
- [ ] TLS 1.2+: Minimum TLS version enforced
- [ ] Certificate Validation: Valid SSL/TLS certificates installed
- [ ] HSTS Enabled: HTTP Strict Transport Security headers configured
- [ ] Secure Cookies: Better Auth cookies use
Secureflag
Implementation:
# FastAPI main.py
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
app = FastAPI()
# Redirect HTTP to HTTPS (production only)
if os.getenv("ENVIRONMENT") == "production":
app.add_middleware(HTTPSRedirectMiddleware)// Next.js Better Auth config
export const auth = betterAuth({
advanced: {
useSecureCookies: process.env.NODE_ENV === "production"
}
})🎯 2. JWT Validation
- [ ] Signature Verification: Always verify JWT signature
- [ ] Expiration Check: Validate
expclaim (reject expired tokens) - [ ] Issuer Validation: Verify
issclaim matches Better Auth URL - [ ] Audience Validation: Verify
audclaim matches expected value - [ ] Algorithm Whitelist: Only allow EdDSA (no HS256, RS256, etc.)
- [ ] Token Replay Protection: Consider adding
jti(JWT ID) for one-time use
Implementation:
payload = jwt.decode(
token,
signing_key,
algorithms=["EdDSA"], # Whitelist only EdDSA
audience=JWT_AUDIENCE,
issuer=JWT_ISSUER,
options={
"verify_signature": True, # ✅ Required
"verify_exp": True, # ✅ Required
"verify_aud": True, # ✅ Required
"verify_iss": True, # ✅ Required
}
)🛡️ 3. User Isolation & Authorization
- [ ] User ID Verification: Always verify
user_idfrom JWT matches URL - [ ] Database Filtering: Filter queries by authenticated user
- [ ] Authorization Middleware: Use FastAPI dependencies for checks
- [ ] Row-Level Security: Database enforces user isolation (if applicable)
- [ ] No Direct Object Reference: Don't accept arbitrary IDs from clients
Implementation:
async def verify_user_access(
user_id: str,
current_user: Dict = Depends(get_current_user)
) -> Dict:
"""Verify user can only access their own resources."""
if current_user["user_id"] != user_id:
raise HTTPException(
status_code=403,
detail="Not authorized to access this resource"
)
return current_user
@router.get("/{user_id}/tasks")
async def get_tasks(
user_id: str,
user: dict = Depends(verify_user_access) # ✅ Enforced
):
# Safe: user_id verified to match authenticated user
return db.query(Task).filter(Task.user_id == user_id).all()🔑 4. Secrets Management
- [ ] Environment Variables: Secrets stored in env vars (not code)
- [ ] Secret Rotation: Plan for rotating BETTER_AUTH_SECRET
- [ ] Secret Length: BETTER_AUTH_SECRET is at least 32 characters
- [ ] No Hardcoded Secrets: No secrets committed to version control
- [ ] Production Secrets: Different secrets for dev/staging/production
Implementation:
# .env (never commit this file)
BETTER_AUTH_SECRET="<min-32-chars-randomly-generated>"
BETTER_AUTH_URL="https://your-domain.com"
DATABASE_URL="postgresql://..."
# .gitignore
.env
.env.local
.env.production🌐 5. CORS Configuration
- [ ] Allowed Origins: Only whitelist trusted domains
- [ ] No Wildcards: Never use
*for allowed origins in production - [ ] Credentials Allowed: Set
allow_credentials=Truefor cookies/auth - [ ] Methods Whitelist: Only allow necessary HTTP methods
- [ ] Headers Whitelist: Only allow necessary headers
Implementation:
from fastapi.middleware.cors import CORSMiddleware
# Production: Specific origins only
allowed_origins = os.getenv("ALLOWED_ORIGINS", "").split(",")
app.add_middleware(
CORSMiddleware,
allow_origins=allowed_origins, # ✅ Specific domains
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
allow_headers=["Authorization", "Content-Type"],
)
# NOT: allow_origins=["*"] # ❌ Never in production🚨 6. Error Handling & Logging
- [ ] Generic Error Messages: Don't leak sensitive info to users
- [ ] Detailed Logging: Log auth failures with details (server-side only)
- [ ] Log Monitoring: Set up alerts for authentication anomalies
- [ ] Rate Limiting: Limit login attempts per IP
- [ ] Audit Trail: Log all authentication events
Implementation:
import logging
logger = logging.getLogger(__name__)
try:
payload = verify_jwt_token(token)
except Exception as e:
# ✅ Generic message to user
raise HTTPException(
status_code=401,
detail="Could not validate credentials"
)
# ✅ Detailed logging (server-side)
logger.error(f"JWT verification failed: {e}", extra={
"token_kid": unverified_header.get("kid"),
"error_type": type(e).__name__,
"client_ip": request.client.host
})⏱️ 7. Token Expiration & Refresh
- [ ] Short Expiration: Tokens expire in reasonable time (e.g., 7 days)
- [ ] Refresh Mechanism: Frontend refreshes tokens before expiry
- [ ] Session Management: Better Auth handles session renewal
- [ ] Logout Support: Implement proper logout (clear tokens)
Implementation:
// Better Auth config
export const auth = betterAuth({
plugins: [
jwt({
expiresIn: "7d" // ✅ Reasonable expiration
})
]
})
// Frontend: Auto-refresh before expiry
useEffect(() => {
const checkSession = async () => {
const session = await authClient.getSession()
if (session && isTokenExpiringSoon(session.token)) {
await authClient.refreshSession()
}
}
const interval = setInterval(checkSession, 60000) // Check every minute
return () => clearInterval(interval)
}, [])🔍 8. JWKS Security
- [ ] JWKS Endpoint Public: Accessible without authentication (standard)
- [ ] Cache Strategy: JWKS responses cached appropriately
- [ ] Key Rotation: System handles key rotation gracefully
- [ ] HTTPS for JWKS: JWKS endpoint uses HTTPS
- [ ] Timeout Configured: JWKS fetch has reasonable timeout
Implementation:
@lru_cache(maxsize=1)
def get_jwks() -> Dict[str, Any]:
try:
response = httpx.get(
JWKS_URL,
timeout=5.0 # ✅ Fail fast on timeout
)
response.raise_for_status()
return response.json()
except httpx.HTTPError:
# ✅ Graceful degradation
raise HTTPException(
status_code=503,
detail="Authentication service unavailable"
)🛠️ 9. Dependency Security
- [ ] Dependency Scanning: Regular scans for vulnerabilities (e.g., Snyk, Dependabot)
- [ ] Version Pinning: Lock dependency versions in production
- [ ] Security Updates: Process for applying security patches
- [ ] Minimal Dependencies: Only install necessary packages
Implementation:
# requirements.txt - Pin versions
fastapi==0.109.0
python-jose[cryptography]==3.3.0
pyjwt==2.8.0
cryptography==41.0.7
httpx==0.25.2
# Run security audit
pip-audit
# GitHub Dependabot
# .github/dependabot.yml
version: 2
updates:
- package-ecosystem: "pip"
directory: "/backend"
schedule:
interval: "weekly"📊 10. Monitoring & Alerts
- [ ] Auth Failure Monitoring: Track authentication failures
- [ ] JWKS Availability: Monitor JWKS endpoint health
- [ ] Token Validation Latency: Track verification performance
- [ ] Anomaly Detection: Alert on unusual auth patterns
- [ ] Error Rate Alerts: Alert on spike in auth errors
Implementation:
from prometheus_client import Counter, Histogram
# Metrics
auth_attempts = Counter("auth_attempts_total", "Total authentication attempts", ["status"])
auth_latency = Histogram("auth_verification_seconds", "JWT verification latency")
@auth_latency.time()
def verify_jwt_token(token: str) -> Dict[str, Any]:
try:
payload = jwt.decode(...)
auth_attempts.labels(status="success").inc()
return payload
except Exception as e:
auth_attempts.labels(status="failure").inc()
raiseSecurity Testing Checklist
Penetration Testing
- [ ] Token Tampering: Modify token payload/signature (should fail)
- [ ] Expired Tokens: Use old tokens (should fail)
- [ ] Token Reuse: Replay captured tokens (should work but limited by expiry)
- [ ] Algorithm Confusion: Change token algorithm in header (should fail)
- [ ] Cross-User Access: Try accessing other users' resources (should fail 403)
- [ ] JWKS Manipulation: Test with modified JWKS response (should fail)
Load Testing
- [ ] JWKS Cache: Verify caching under load
- [ ] Auth Performance: Measure verification latency
- [ ] Rate Limiting: Test rate limit effectiveness
Incident Response Plan
If Token Compromise Suspected:
1. Rotate BETTER_AUTH_SECRET (forces all tokens invalid) 2. Force user re-authentication (clear all sessions) 3. Review access logs for unauthorized access 4. Update JWKS keys (Better Auth handles this) 5. Notify affected users if data exposed
If JWKS Endpoint Compromised:
1. Better Auth compromise - rotate all keys immediately 2. Man-in-the-middle - verify HTTPS configuration 3. DNS poisoning - check DNS records
Compliance Considerations
GDPR / Data Privacy
- [ ] User Consent: Obtain consent for authentication
- [ ] Data Minimization: JWT contains only necessary claims
- [ ] Right to Erasure: Implement user deletion
- [ ] Data Portability: Allow users to export their data
SOC 2 / ISO 27001
- [ ] Access Controls: Documented and enforced
- [ ] Audit Logging: Authentication events logged
- [ ] Encryption in Transit: TLS for all communications
- [ ] Key Management: Documented secret rotation process
Quick Security Audit
Run this checklist before every deployment:
# 1. Check HTTPS enforcement
curl http://your-api.com/api/v1/test # Should redirect to HTTPS
# 2. Verify JWKS endpoint
curl https://your-auth.com/api/auth/jwks # Should return public keys
# 3. Test expired token rejection
curl -H "Authorization: Bearer <expired-token>" \
https://your-api.com/api/v1/test # Should return 401
# 4. Test cross-user access
curl -H "Authorization: Bearer <user1-token>" \
https://your-api.com/api/v1/user2/tasks # Should return 403
# 5. Check security headers
curl -I https://your-api.com # Should include security headersAdditional Resources
Troubleshooting Guide
Common issues and solutions for Better Auth + FastAPI JWT integration.
Table of Contents
1. Database Schema Issues 2. JWKS Endpoint Issues 3. Token Verification Failures 4. User Authorization Errors 5. Frontend Integration Issues 6. Better Auth UUID Integration Issues 7. Performance Problems 8. Development vs Production Issues
---
Database Schema Issues
Issue: "column 'token' of relation 'session' does not exist"
Symptoms:
ERROR [Better Auth]: column "token" of relation "session" does not exist
POST /api/auth/sign-up/email 500 in 1255msCause: The session table is missing the required token column. This is a core Better Auth field, not specific to the JWT plugin.
Why This Happens:
- Initial migration was created without the
tokencolumn - Better Auth requires this field to store the session token (used as cookie value)
- Common issue when manually creating Better Auth tables instead of using generated migrations
Solution:
Create a migration to add the token column:
# alembic revision -m "add_token_column_to_session_table"
def upgrade() -> None:
# Add token column to session table
op.add_column('session', sa.Column('token', sa.String(), nullable=False, server_default=''))
# Remove server_default after adding column (it's just for the migration)
op.alter_column('session', 'token', server_default=None)
def downgrade() -> None:
# Remove token column from session table
op.drop_column('session', 'token')Apply the migration:
# Using Alembic
alembic upgrade head
# Or using Better Auth CLI (Next.js)
npx @better-auth/cli migrateVerify the fix:
-- Check session table schema
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = 'session'
ORDER BY ordinal_position;Expected columns:
id(character varying, NOT NULL)userId(character varying, NOT NULL)expiresAt(timestamp, NOT NULL)ipAddress(character varying, nullable)userAgent(character varying, nullable)createdAt(timestamp, NOT NULL)updatedAt(timestamp, NOT NULL)token(character varying, NOT NULL) ✅
Issue: "relation 'jwks' does not exist"
Symptoms:
ERROR [Better Auth]: relation "jwks" does not exist
POST /api/auth/sign-up/email 200 in 2.9sUser signs up/logs in successfully but then gets redirected to login page instead of dashboard.
Cause: JWT plugin enabled but the jwks table was not created in the database. Better Auth needs this table to store public/private key pairs for JWT signature verification.
Why This Happens:
- JWT plugin was added to Better Auth config
- Better Auth CLI migration was not run, OR
- Backend database migrations were created manually and forgot to include JWKS table
Solution (Backend - Alembic Migration):
If using FastAPI with Alembic, create a migration to add the JWKS table:
# Generate new migration file
alembic revision -m "create_jwks_table_for_jwt_plugin"Then edit the migration file:
# alembic/versions/xxxxx_create_jwks_table_for_jwt_plugin.py
from alembic import op
import sqlalchemy as sa
def upgrade() -> None:
"""Create JWKS table for Better Auth JWT plugin."""
op.create_table(
'jwks',
sa.Column('id', sa.String(), nullable=False),
sa.Column('publicKey', sa.String(), nullable=False),
sa.Column('privateKey', sa.String(), nullable=False),
sa.Column('createdAt', sa.DateTime(), nullable=False, server_default=sa.text('NOW()')),
sa.Column('expiresAt', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
def downgrade() -> None:
"""Drop JWKS table."""
op.drop_table('jwks')Apply the migration:
alembic upgrade headSolution (Frontend - Better Auth CLI):
If using Better Auth CLI for database management (Next.js only):
# Next.js frontend (Better Auth CLI)
npx @better-auth/cli migrate
# Or generate schema
npx @better-auth/cli generateVerify the fix:
-- Check if jwks table exists
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'public'
AND table_name = 'jwks';Expected result: One row with table_name = 'jwks'
Verify columns:
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = 'jwks'
ORDER BY ordinal_position;Expected columns:
id(character varying, NOT NULL) ✅publicKey(character varying, NOT NULL) ✅privateKey(character varying, NOT NULL) ✅createdAt(timestamp, NOT NULL) ✅expiresAt(timestamp, nullable) ✅
Important: The JWT plugin creates the jwks table, but does NOT modify the session table. The token column issue is separate from JWT plugin configuration.
---
JWKS Endpoint Issues
Issue: "Unable to fetch JWKS" or "JWKS endpoint not accessible"
Symptoms:
HTTPException 503: Unable to fetch authentication keysCauses & Solutions:
1. Better Auth not running
# Verify Better Auth is running
curl http://localhost:3000/api/auth/jwksSolution: Start your Next.js application with Better Auth
2. Wrong JWKS URL
# Check BETTER_AUTH_URL environment variable
echo $BETTER_AUTH_URLSolution: Ensure BETTER_AUTH_URL matches your Next.js URL
3. Network connectivity
# Test connectivity from FastAPI container
docker exec -it fastapi-container curl http://nextjs:3000/api/auth/jwksSolution: Check Docker networks, firewalls, or DNS resolution
4. CORS blocking (frontend to JWKS) Solution: JWKS endpoint should not require CORS (server-to-server)
Issue: "JWKS response missing 'keys' field"
Symptoms:
ValueError: JWKS response missing 'keys' fieldCause: JWT plugin not enabled or database migration not run
Solution:
// 1. Verify JWT plugin in auth config
export const auth = betterAuth({
plugins: [jwt()] // ✅ Must be present
})
// 2. Run database migration
npm run db:migrate---
Token Verification Failures
Issue: "Unable to find matching signing key"
Symptoms:
HTTPException 401: Unable to find matching signing keyCauses & Solutions:
1. Key rotation occurred
# Cache invalidation is automatic, but you can manually clear:
from app.auth.jwt_verification import get_jwks
get_jwks.cache_clear()2. Token from different issuer
# Decode token to check issuer (without verification)
python -c "import jwt; print(jwt.decode('TOKEN', options={'verify_signature': False}))"Solution: Ensure token is from correct Better Auth instance
3. Development vs production mismatch
- Token from dev Better Auth, but backend expects production
- Solution: Match environments or use separate tokens
Issue: "Token has expired"
Symptoms:
HTTPException 401: Token has expiredCause: Token's exp claim is in the past
Solutions:
1. Frontend session refresh
// Add auto-refresh logic
const session = await authClient.getSession()
if (session && isExpiringSoon(session.token)) {
await authClient.refreshSession()
}2. Increase token expiration (not recommended for security)
export const auth = betterAuth({
plugins: [
jwt({ expiresIn: "30d" }) // Default is 7d
]
})3. Check server time sync
# Ensure servers have synchronized time
date -u # Should match across all serversIssue: "Invalid token claims" (aud/iss mismatch)
Symptoms:
HTTPException 401: Invalid token claimsCause: Audience or issuer doesn't match expected values
Solution:
# backend/.env - Must match Better Auth URL exactly
BETTER_AUTH_URL="http://localhost:3000" # Dev
BETTER_AUTH_URL="https://your-domain.com" # Prod
# frontend/.env.local
BETTER_AUTH_URL="http://localhost:3000" # Must match backendDebug:
# Decode token to see actual aud/iss
import jwt
payload = jwt.decode(token, options={"verify_signature": False})
print(f"Issuer: {payload.get('iss')}")
print(f"Audience: {payload.get('aud')}")Issue: "Signature verification failed"
Symptoms:
jwt.JWTError: Signature verification failedCauses & Solutions:
1. Token tampered with
- Solution: Token is invalid, user must re-authenticate
2. Wrong algorithm
# Ensure algorithm whitelist is correct
algorithms=["EdDSA"] # Better Auth uses Ed255193. Public key mismatch
- Verify JWKS contains the correct public key
curl http://localhost:3000/api/auth/jwks---
User Authorization Errors
Issue: 403 Forbidden - "Not authorized to access this resource"
Symptoms:
HTTPException 403: Not authorized to access this user's resourcesCause: user_id in URL doesn't match authenticated user
Debug:
# Check what's being compared
print(f"URL user_id: {user_id}")
print(f"Token user_id: {current_user['user_id']}")Solutions:
1. Frontend using wrong user_id
// ✅ Use authenticated user's ID
const session = await authClient.getSession()
const tasks = await getTasks(session.user.id)
// ❌ NOT hardcoded or from URL params
const tasks = await getTasks("some-other-user-id")2. Token has wrong user_id
- Verify token payload:
# Decode token to check sub claim
python -c "import jwt; print(jwt.decode('TOKEN', options={'verify_signature': False})['sub'])"Issue: Users seeing each other's data
Critical Security Issue!
Symptoms: User A can access User B's tasks/data
Root Cause: Missing authorization check
Solution:
# ❌ WRONG - No authorization
@router.get("/{user_id}/tasks")
async def get_tasks(user_id: str):
return db.query(Task).filter(Task.user_id == user_id).all()
# ✅ CORRECT - With authorization
@router.get("/{user_id}/tasks")
async def get_tasks(
user_id: str,
user: dict = Depends(verify_user_access) # Required!
):
return db.query(Task).filter(Task.user_id == user_id).all()---
Frontend Integration Issues
Issue: "authClient.useSession is not a function"
Symptoms:
TypeError: authClient.useSession is not a functionCause: Better Auth does not export a useSession() React hook. The correct API is authClient.getSession() which is an async function, not a hook.
Wrong Code:
// ❌ WRONG - This doesn't exist in Better Auth
const { data: session } = authClient.useSession()Solution:
Use authClient.getSession() with React's useEffect hook:
// ✅ CORRECT - Load session in useEffect
import { useState, useEffect } from "react"
import { authClient } from "@/lib/auth-client"
function MyComponent() {
const [user, setUser] = useState(null)
const [isLoading, setIsLoading] = useState(true)
useEffect(() => {
async function loadSession() {
try {
const session = await authClient.getSession()
if (session?.data?.user) {
setUser(session.data.user)
} else {
// No session - redirect to login
router.push("/auth/login")
}
} catch (error) {
console.error("Failed to load session:", error)
} finally {
setIsLoading(false)
}
}
loadSession()
}, [])
if (isLoading) {
return <div>Loading...</div>
}
return <div>Welcome {user?.name}</div>
}Key Points:
authClient.getSession()is async - must use withawait- Call it inside
useEffectfor React components - Add loading state to prevent rendering before session loads
- Handle redirect to login if no session exists
Issue: "Authorization header missing"
Symptoms: Backend receives request without Authorization header
Causes & Solutions:
1. Forgot to include header
// ❌ WRONG
fetch('/api/v1/user123/tasks')
// ✅ CORRECT
const token = session.session.token
fetch('/api/v1/user123/tasks', {
headers: {
'Authorization': `Bearer ${token}`
}
})2. Token not available
// Check if session exists
const session = await authClient.getSession()
if (!session) {
router.push('/login') // Redirect to login
return
}Issue: "Token is null or undefined" / "No authentication token available"
Symptoms:
TypeError: Cannot read property 'token' of null
Error: No authentication token available. Please log in.Cause: 1. User not authenticated or session expired 2. Incorrect session path - using session.session.token instead of session.data.session.token
Solution:
⚠️ IMPORTANT: Better Auth client returns session data in session.data, not directly in session.
// ❌ WRONG - Common mistake
const session = await authClient.getSession()
if (!session?.session) { // Wrong path!
return null
}
const token = session.session.token // Wrong path!
// ✅ CORRECT - Use session.data.session
const session = await authClient.getSession()
if (!session?.data?.session) { // Correct path
console.error("No valid session found")
router.push("/auth/login")
return
}
const token = session.data.session.token // Correct pathComplete Example:
async function getAuthToken(): Promise<string | null> {
const session = await authClient.getSession()
// Check session.data.session (not session.session!)
if (!session?.data?.session) {
return null
}
// Extract token from session.data.session.token
return session.data.session.token
}Issue: CORS errors from frontend
Symptoms:
Access to fetch at 'http://localhost:8000/api/v1/tasks' from origin
'http://localhost:3000' has been blocked by CORS policySolution:
# backend/main.py
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"], # Frontend URL
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)---
Better Auth UUID Integration Issues
Issue: "Token missing UUID (uuid claim)"
Symptoms:
HTTPException 401: Token missing UUID (uuid claim)Cause: Better Auth hook not configured to fetch UUID or JWT plugin not including UUID in custom claims.
Why This Happens:
- Better Auth uses String IDs by default (
subclaim) - Application needs UUID for type consistency
- UUID must be added via custom claim in JWT payload
Solution:
1. Add UUID column to user table (database migration):
-- Add UUID column with auto-generation
ALTER TABLE "user"
ADD COLUMN uuid UUID UNIQUE NOT NULL DEFAULT gen_random_uuid();
-- Create index for performance
CREATE INDEX idx_user_uuid ON "user"(uuid);2. Configure Better Auth hook (frontend):
// lib/auth.ts
import { Pool } from "pg"
const pool = new Pool({ connectionString: process.env.DATABASE_URL })
export const auth = betterAuth({
hooks: {
user: {
created: async ({ user }) => {
// Fetch UUID generated by database
const result = await pool.query(
'SELECT uuid FROM "user" WHERE id = $1',
[user.id]
)
const uuid = result.rows[0]?.uuid
return { ...user, uuid }
}
}
},
plugins: [
jwt({
algorithm: "EdDSA",
async jwt(user, session) {
return {
uuid: user.uuid, // Include UUID in JWT ⭐
}
},
}),
],
})3. Update backend to extract UUID:
# backend/app/auth/dependencies.py
from uuid import UUID
async def get_current_user(token: str = Depends(oauth2_scheme)):
payload = verify_jwt_token(token)
# Extract UUID from custom claim (not 'sub')
user_uuid_str = payload.get("uuid") # ⭐
if not user_uuid_str:
raise HTTPException(401, "Token missing UUID claim")
user_uuid = UUID(user_uuid_str)
# Query by UUID
user = await session.execute(
select(User).where(User.uuid == user_uuid)
)
return user.scalar_one_or_none()Verification:
# Decode JWT to verify UUID is present
python -c "
import jwt
token = 'YOUR_JWT_TOKEN'
payload = jwt.decode(token, options={'verify_signature': False})
print('UUID claim:', payload.get('uuid'))
"Issue: User not found after registration (dual auth system conflict)
Symptoms:
- User registers successfully via frontend
- Backend returns 401 "User not found"
- Database has
usertable (Better Auth) anduserstable (custom)
Cause: Two conflicting authentication systems - Better Auth uses user table, backend queries users table.
Solution: Use Better Auth as single source of truth with UUID extension (hybrid ID approach).
Implementation:
1. Database Migration (3 steps):
# Migration 1: Add UUID to Better Auth user table
def upgrade():
op.add_column('user', sa.Column('uuid', sa.UUID(), nullable=False,
server_default=sa.text('gen_random_uuid()')))
op.create_unique_constraint('uq_user_uuid', 'user', ['uuid'])
op.create_index('idx_user_uuid', 'user', ['uuid'])
# Migration 2: Update foreign keys
def upgrade():
# Point all FKs to user.uuid (not users.id)
op.drop_constraint('tasks_user_id_fkey', 'tasks', type_='foreignkey')
op.create_foreign_key('tasks_user_uuid_fkey', 'tasks', 'user',
['user_id'], ['uuid'], ondelete='CASCADE')
# Migration 3: Drop custom users table
def upgrade():
op.drop_table('users', if_exists=True)2. Backend Model (map to Better Auth schema):
# backend/models/user.py
class User(SQLModel, table=True):
__tablename__ = "user" # Better Auth table (singular!)
# Better Auth fields
id: str = Field(primary_key=True)
email: str = Field(unique=True, index=True)
emailVerified: bool = Field(default=False)
name: Optional[str] = None
createdAt: datetime
updatedAt: datetime
# Application field
uuid: UUID = Field(unique=True, index=True, nullable=False)3. Remove backend auth endpoints - Better Auth handles registration/login on frontend.
Key Pattern: Always query by User.uuid and validate against UUID from JWT custom claim.
Issue: UUID vs String ID mismatch in user isolation
Symptoms:
HTTPException 403: Not authorized to access this user's resourcesCause: Comparing UUID from JWT with String ID from URL, or vice versa.
Solution: Ensure consistent UUID usage:
# ❌ Wrong - comparing String and UUID
if current_user["user_id"] != user_id: # user_id is UUID, user_id is String
# ✅ Correct - comparing UUIDs
from uuid import UUID
current_user_uuid = UUID(current_user["uuid"])
if current_user_uuid != user_id: # Both are UUIDAPI Route Pattern:
from uuid import UUID
@router.get("/{user_id}/tasks")
async def get_tasks(
user_id: UUID, # ⭐ UUID in path
user: dict = Depends(verify_user_access)
):
# user_id validated against JWT UUID
return get_user_tasks(user_id)---
Performance Problems
Issue: Slow authentication (every request fetches JWKS)
Symptoms: High latency on authenticated requests
Cause: JWKS caching not working
Solution:
# Verify @lru_cache is present
@lru_cache(maxsize=1) # ✅ Must have this
def get_jwks() -> Dict[str, Any]:
# ...Monitor cache hits:
import functools
# Check cache info
print(get_jwks.cache_info())
# CacheInfo(hits=100, misses=1, maxsize=1, currsize=1)Issue: JWKS fetch timeout
Symptoms:
httpx.ReadTimeout: Read operation timed outCauses & Solutions:
1. Better Auth server slow/down
- Check Better Auth server health
- Increase timeout (temporarily):
response = httpx.get(JWKS_URL, timeout=10.0) # Increase from 5.02. Network latency
- Deploy FastAPI and Next.js in same region/network
- Use internal network addresses in Docker/Kubernetes
---
Development vs Production Issues
Issue: Works locally but fails in production
Common Causes:
1. HTTP vs HTTPS
# Local (HTTP)
BETTER_AUTH_URL="http://localhost:3000"
# Production (HTTPS)
BETTER_AUTH_URL="https://your-domain.com"2. Environment variables not set
# Check production environment variables
printenv | grep BETTER_AUTH3. CORS configuration
# Development
allow_origins=["http://localhost:3000"]
# Production
allow_origins=["https://your-domain.com"]4. Database connection
- Verify DATABASE_URL is correct for production
Issue: Works in production but fails locally
Common Causes:
1. Docker networking
# docker-compose.yml
services:
nextjs:
networks:
- app-network
fastapi:
environment:
- BETTER_AUTH_URL=http://nextjs:3000 # Use service name
networks:
- app-network2. Port conflicts
- Check if ports 3000 (Next.js) and 8000 (FastAPI) are available
---
Debugging Tools
1. Decode JWT (without verification)
# Using Python
python3 << 'EOF'
import jwt
import sys
token = "YOUR_TOKEN_HERE"
payload = jwt.decode(token, options={"verify_signature": False})
import json
print(json.dumps(payload, indent=2))
EOF2. Test JWKS endpoint
# Verify JWKS is accessible
curl -s http://localhost:3000/api/auth/jwks | jq .
# Check specific fields
curl -s http://localhost:3000/api/auth/jwks | jq '.keys[0].kid'3. Test token verification
# Use test script
python scripts/test_jwt_verification.py \
--jwks-url http://localhost:3000/api/auth/jwks \
--token "YOUR_TOKEN"4. Enable debug logging
# backend/main.py
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("app.auth")
logger.setLevel(logging.DEBUG)5. Network diagnostics
# Test connectivity
ping nextjs-container
curl -v http://nextjs:3000/api/auth/jwks
# Check DNS resolution
nslookup your-domain.com
# Test from within container
docker exec -it fastapi-container bash
curl http://nextjs:3000/api/auth/jwks---
Emergency Procedures
If authentication is completely broken:
1. Check Better Auth is running
curl http://localhost:3000/health2. Verify JWT plugin enabled
// lib/auth.ts
plugins: [jwt()] // Must be present3. Clear all caches
get_jwks.cache_clear()4. Restart all services
docker-compose restart5. Check logs for errors
docker-compose logs -f fastapi
docker-compose logs -f nextjsIf users are locked out:
1. Extend token expiration (temporary fix)
jwt({ expiresIn: "30d" })2. Force re-authentication (clear sessions)
- Have users log out and log back in
3. Verify HTTPS in production
- Tokens may not work over HTTP in production
---
Getting Help
If none of these solutions work:
1. Enable debug logging and check logs 2. Use test scripts to isolate the issue 3. Verify environment variables match across services 4. Check Better Auth documentation for updates 5. Review FastAPI logs for detailed error messages
Useful commands:
# View all environment variables
docker exec -it fastapi-container env | grep BETTER_AUTH
# Test JWT verification in Python REPL
docker exec -it fastapi-container python
>>> from app.auth.jwt_verification import verify_jwt_token
>>> verify_jwt_token("YOUR_TOKEN")
# Monitor requests
docker logs -f fastapi-container | grep "JWT"---
Frontend-Backend Integration Issues
Updated: 2026-01-02
This section documents critical issues encountered when integrating Next.js frontend with FastAPI backend after authentication is working. These are data flow and type alignment issues, not authentication problems.
Issue: Tasks not displaying despite successful API response
Symptoms:
- Backend returns 200 OK with task data
- Network tab shows data being received
- Dashboard remains empty (no tasks displayed)
Root Cause: Backend returns plain array List[TaskResponse] but frontend expected paginated response {items: [], total: 0, total_pages: 0}
Why This Happens:
- Backend API returns
List[TaskResponse]directly - Frontend code assumes
response.itemsexists response.itemsevaluates toundefined- Tasks state set to empty array despite data present
Solution:
Handle both response formats defensively:
// frontend/src/contexts/TaskContext.tsx
const refreshTasks = async () => {
const response = await apiClient.get(`/api/v1/${userId}/tasks`);
// Handle both array response (current backend) and paginated response (future)
if (Array.isArray(response)) {
// Backend returns plain array
setTasks(response);
setTotalTasks(response.length);
setTotalPages(1);
setCurrentPage(1);
setPageLimit(50);
} else {
// Backend returns paginated response (future implementation)
setTasks(response.items || []);
setTotalTasks(response.total || 0);
setTotalPages(response.total_pages || 0);
setCurrentPage(response.page || 1);
setPageLimit(response.limit || 50);
}
};Prevention: 1. Always check backend response structure in API contracts before coding 2. Read Pydantic response models (backend/src/schemas/task.py) 3. Add defensive checks for both current and future response formats 4. Test with actual backend, not mocked data
---
Issue: Tag filtering crashes at runtime
Symptoms:
TypeError: Cannot read property 'includes' of undefinedRoot Cause: Frontend expected tags to be array of IDs (number[]) but backend returns full tag objects {id, name, color}
Why This Happens:
- Backend Pydantic schema:
tags: List[TagResponse](full objects) - Frontend TypeScript type:
tags: number[](just IDs) - Code tried
t.tags.includes(tagId)expecting array of numbers - Actual data: array of objects,
includes()fails
Wrong Code:
// ❌ WRONG - assumes tags are primitive values
if (selectedTags.length > 0) {
filtered = filtered.filter((t) =>
t.tags.includes(tag) // Fails: comparing objects
);
}Solution:
// ✅ CORRECT - handle tag objects
if (selectedTags.length > 0) {
filtered = filtered.filter((t) =>
Array.isArray(t.tags) && selectedTags.some((tagId) =>
t.tags.some((tag) => tag.id.toString() === tagId || tag.name === tagId)
)
);
}Align TypeScript types with backend:
// frontend/src/types/task-schema.ts
export interface Task {
// ...
tags: Array<{id: number, name: string, color?: string}> // ✅ Match backend
// NOT: tags: number[] // ❌ Wrong
}Prevention: 1. Always read backend schemas first (backend/src/schemas/) 2. Align TypeScript interfaces with Pydantic models exactly 3. Never assume array element types without verification 4. Check TaskResponse and TagResponse schemas
Quick Check:
# View backend schema
cat backend/src/schemas/task.py | grep "tags:"
# tags: List[TagResponse] = Field(default_factory=list)
# Update frontend type to match---
Issue: Priority sorting showing "NaN" in pagination
Symptoms:
- Pagination displays "Page NaN of NaN"
- Tasks don't sort by priority correctly
Root Cause: Priority is optional field (Optional[PriorityEnum]) but code didn't handle undefined
Wrong Code:
// ❌ WRONG - arithmetic on undefined gives NaN
case "priority":
const priorityOrder = { high: 3, medium: 2, low: 1 };
comparison = priorityOrder[b.priority] - priorityOrder[a.priority];
break;Solution:
// ✅ CORRECT - null checks with defaults
case "priority":
const priorityOrder = { high: 3, medium: 2, low: 1 };
const aPriority = a.priority ? priorityOrder[a.priority] : 0;
const bPriority = b.priority ? priorityOrder[b.priority] : 0;
comparison = bPriority - aPriority;
break;Prevention: 1. Check Pydantic schema for Optional[Type] fields 2. Add null/undefined checks before accessing optional fields 3. Provide sensible defaults (0 for numeric comparisons) 4. Test with data that has missing optional fields
Quick Check:
# Find optional fields in backend schema
grep "Optional\[" backend/src/schemas/task.py
# priority: Optional[PriorityEnum] = None ⚠️ Can be None---
Issue: Tags not inserting into task_tags table
Symptoms:
- Create task with tags selected
- Task created successfully
- Database has no records in
task_tagstable - Tags don't appear on task cards
Root Cause: Backend TaskCreate schema doesn't accept tags field - tags must be assigned via separate endpoint
Wrong Code:
// ❌ WRONG - backend doesn't accept tags in TaskCreate
await addTask({
title: "Buy groceries",
tags: [1, 2, 3] // Ignored by backend!
});Solution - Multi-step operation:
// ✅ CORRECT - create task, then assign tags
// Step 1: Create task (without tags)
const createdTask = await addTask({
title: "Buy groceries",
description: "Milk, eggs",
tags: [] // Empty or omit
});
// Step 2: Assign tags via separate API calls
if (data.tags && data.tags.length > 0) {
for (const tagId of data.tags) {
try {
await apiClient.post(`/api/v1/${userId}/tasks/${createdTask.id}/tags`, {
tag_id: tagId,
});
} catch (tagError) {
console.error(`Failed to assign tag ${tagId}:`, tagError);
// Continue with other tags even if one fails
}
}
}
// Step 3: Refresh to show tags
await refreshTasks();Backend Endpoints:
# backend/src/api/tasks.py
POST /api/v1/{user_id}/tasks # Create task (no tags field)
POST /api/v1/{user_id}/tasks/{id}/tags # Assign tag to task
DELETE /api/v1/{user_id}/tasks/{id}/tags/{tag_id} # Remove tagPrevention: 1. Read API endpoint documentation 2. Check which fields are accepted in create/update schemas 3. Understand multi-step operations (create entity, then add associations) 4. Look for separate endpoints for associations
---
Issue: Edit form not pre-filling reminder time, recurrence, and tags
Symptoms:
- Open edit task modal
- Title and description populate correctly
- Reminder time, recurrence dropdown, and tags checkboxes are blank
- Values exist in task object
Root Causes (3 separate issues):
1. Uncontrolled vs Controlled Components
Select components using defaultValue don't update when props change:
// ❌ WRONG - uncontrolled, doesn't update
<Select defaultValue={field.value}>
<SelectTrigger>...</SelectTrigger>
</Select>
// ✅ CORRECT - controlled, updates on prop changes
<Select value={field.value} onValueChange={field.onChange}>
<SelectTrigger>...</SelectTrigger>
</Select>2. Field Name Mismatches
Frontend uses different field names than backend:
// ❌ WRONG field names
reminder_time // Frontend
recurrence // Frontend
// ✅ CORRECT field names (match backend)
reminder_at // Backend Pydantic schema
recurrence_pattern // Backend Pydantic schema3. Datetime Format Conversion
Backend ISO 8601 format doesn't work with HTML datetime-local input:
// Backend format: "2024-12-20T10:00:00.000Z"
// HTML datetime-local format: "2024-12-20T10:00"
// ✅ Add conversion helper
const toDatetimeLocal = (isoString?: string) => {
if (!isoString) return "";
try {
// Remove timezone and seconds
return isoString.slice(0, 16); // "2024-12-20T10:00"
} catch {
return "";
}
};
// Use in form reset
reminder_at: toDatetimeLocal(task.reminder_at),Complete Fix:
// Form field names must match backend exactly
const formData = {
title: task.title,
description: task.description || "",
priority: task.priority || "medium",
due_date: task.due_date || "",
reminder_at: toDatetimeLocal(task.reminder_at), // Format conversion
recurrence_pattern: task.recurrence_pattern || "none", // Correct name
tags: Array.isArray(task.tags) ? task.tags.map((t) => t.id) : [],
completed: task.completed,
};
form.reset(formData);Prevention: 1. Always use controlled components (value + onChange) for pre-filled forms 2. Verify field names match backend exactly (check Pydantic schemas) 3. Handle datetime format conversions explicitly 4. Test edit mode, not just create mode 5. Extract IDs from objects when needed (tags)
---
Issue: 500 Error - "can't compare offset-naive and offset-aware datetimes"
Symptoms:
TypeError: can't compare offset-naive and offset-aware datetimes
File "backend/src/schemas/task.py", line 64, in reminder_before_due
if reminder >= due:Root Cause: Pydantic validator compared datetimes with different timezone awareness
Why This Happens:
- Frontend sends datetime without timezone:
"2024-12-20T10:00" - Backend receives it as offset-naive datetime
- Due date might be offset-aware (has timezone)
- Python can't compare naive and aware datetimes
Wrong Code:
# ❌ WRONG - no timezone normalization
@field_validator("reminder_at")
@classmethod
def reminder_before_due(cls, v, info):
if v and info.data.get("due_date"):
if v >= info.data["due_date"]: # Fails if timezone mismatch
raise ValueError("reminder_at must be before due_date")
return vSolution:
# ✅ CORRECT - normalize to UTC before comparison
from datetime import timezone
@field_validator("reminder_at")
@classmethod
def reminder_before_due(cls, v: Optional[datetime], info: ValidationInfo) -> Optional[datetime]:
"""Ensure reminder_at is before due_date if both are set."""
if v and info.data.get("due_date"):
due_date = info.data["due_date"]
# Ensure both datetimes are timezone-aware for comparison
reminder = v if v.tzinfo else v.replace(tzinfo=timezone.utc)
due = due_date if due_date.tzinfo else due_date.replace(tzinfo=timezone.utc)
if reminder >= due:
raise ValueError("reminder_at must be before due_date")
return vPrevention: 1. Always normalize timezone awareness before datetime comparisons 2. Use UTC as canonical timezone 3. Test with both timezone-aware and naive datetimes 4. Apply same fix to all datetime validators (TaskCreate, TaskUpdate, TaskReplace)
---
Issue: Tag color validation failures
Symptoms:
- Creating tags with color works
- Creating tags without color fails validation
- Editing tags that have no color fails
Root Cause: Frontend validation required color to be mandatory, backend allows optional
Schema Mismatch:
# Backend (Python) - color is optional
class TagCreate(BaseModel):
name: str
color: Optional[str] = None # ⭐ Optional// Frontend (TypeScript) - color was required
export const createTagSchema = z.object({
name: z.string()...,
color: z.string().regex(...), // ❌ Required (missing .optional())
})Solution:
1. Make color optional in frontend validation:
// ✅ CORRECT - match backend
export const createTagSchema = z.object({
name: z.string()
.min(1, "Tag name is required")
.max(50, "Tag name must be 50 characters or less"),
color: z.string()
.regex(/^#[0-9A-Fa-f]{6}$/, "Must be hex color")
.transform((val) => val.toUpperCase())
.optional(), // ⭐ Match backend
})2. Provide default colors in UI:
// When editing tag without color
color: initialData.color || "#3B82F6", // Default to blue
// When displaying tag without color
style={{ backgroundColor: tag.color || "#3B82F6" }}Prevention: 1. Check backend for Optional[Type] fields 2. Make frontend validation schemas match exactly 3. Provide sensible defaults for optional visual properties 4. Test with data that has missing optional fields
---
Issue: Tag filter checkboxes not working
Symptoms:
- Click tag checkbox in filter dropdown
- Checkbox doesn't check/uncheck
- Tag filtering doesn't work
Root Cause: Type mismatch - backend returns tag IDs as number, FilterContext uses string[]
Why This Happens:
- Backend:
id: int(Python integer) - TypeScript receives:
id: number - FilterContext:
selectedTags: string[] - Checkbox checked logic:
selectedTags.includes(tag.id)→false(because1 !== "1")
Wrong Code:
// ❌ WRONG - number vs string comparison fails
<Checkbox
checked={selectedTags.includes(tag.id)} // tag.id is number, selectedTags has strings
onCheckedChange={(checked) => {
if (checked) {
setSelectedTags([...selectedTags, tag.id]); // Adds number to string[]
}
}}
/>Solution:
// ✅ CORRECT - convert to string for comparison
<Checkbox
checked={selectedTags.includes(tag.id.toString()) || selectedTags.includes(tag.id as any)}
onCheckedChange={(checked) => {
if (checked) {
setSelectedTags([...selectedTags, tag.id.toString()]); // Convert to string
} else {
setSelectedTags(
selectedTags.filter((id) => id !== tag.id.toString() && id !== tag.id)
);
}
}}
/>
// Also fix tag lookup in active filters display
const tag = tags.find((t) => t.id.toString() === tagId.toString());Prevention: 1. Choose one ID type (number or string) and use consistently 2. Document ID types explicitly in TypeScript interfaces 3. Convert at boundaries if mixing types 4. Test filter/selection UIs thoroughly with real data
Alternative Solution: Change FilterContext to use number[]:
// If you control the context
const [selectedTags, setSelectedTags] = useState<number[]>([]); // Use number[]---
Key Learnings for Frontend-Backend Integration
1. Schema Alignment is Critical
Never assume - Always read backend Pydantic schemas before writing frontend TypeScript types:
# Always check these files FIRST
backend/src/schemas/task.py
backend/src/schemas/tag.py
backend/src/schemas/user.py2. Handle Optional Fields Properly
Backend Optional[Type] fields require:
- Null/undefined checks before use
- Default values for display
- Proper type guards in TypeScript
// Always check for optional fields
const priority = task.priority || "medium"; // Default
const color = tag.color || "#3B82F6"; // Visual default3. Multi-Step Operations Need Documentation
When backend requires multiple API calls for one logical operation:
- Document this in API contracts
- Handle partial failures gracefully
- Refresh data after multi-step operations
4. Controlled Components for Forms
React forms that need pre-filling must use controlled components:
<Select value={field.value} onValueChange={field.onChange}> // ✅ Controlled
<Select defaultValue={field.value}> // ❌ Uncontrolled - won't update5. Type Consistency Across Boundaries
- Choose number or string for IDs
- Convert explicitly at boundaries if mixing
- Document type choices in interfaces
6. Datetime Handling Checklist
- [ ] Normalize timezone awareness before comparisons
- [ ] Use UTC as canonical timezone
- [ ] Handle format conversions (ISO 8601 ↔ datetime-local)
- [ ] Test with both aware and naive datetimes
7. Test Beyond Happy Path
- [ ] Test create AND edit modes
- [ ] Test with missing optional fields
- [ ] Test multi-step operations with partial failures
- [ ] Test filter/search with real backend data
- [ ] Test with actual backend, not mocked data
8. Defensive Programming
// Handle both current and future response formats
if (Array.isArray(response)) {
// Current format
} else {
// Future paginated format
}
// Always check array existence
if (Array.isArray(task.tags) && task.tags.length > 0) {
// Process tags
}---
Comprehensive Verification Checklist
Before deploying integration, verify:
- [ ] Read all backend Pydantic schemas
- [ ] TypeScript types match Pydantic models exactly
- [ ] Optional fields have null checks and defaults
- [ ] Field names match exactly (no typos:
reminder_atnotreminder_time) - [ ] Datetime conversions implemented where needed
- [ ] Controlled components used for pre-filled forms
- [ ] ID types consistent across contexts
- [ ] Multi-step operations documented and implemented
- [ ] Error handling for partial failures
- [ ] Tested with actual backend responses
- [ ] Tested both create and edit modes
- [ ] Filter and search tested with real data
- [ ] Defensive checks for response formats
---
For Complete Implementation Guide: See /specs/005-frontend-backend-integration/IMPLEMENTATION_FIXES.md for detailed fixes with file paths and line numbers.
Last Updated: 2026-01-02
#!/usr/bin/env python3
"""
Test JWT token verification using JWKS.
Usage:
python test_jwt_verification.py --jwks-url <url> --token <jwt-token>
Example:
python test_jwt_verification.py \
--jwks-url http://localhost:3000/api/auth/jwks \
--token "eyJhbGci..."
"""
import argparse
import json
import httpx
from jose import jwt, jwk, JWTError
from typing import Dict, Any
def get_jwks(jwks_url: str) -> Dict[str, Any]:
"""Fetch JWKS from Better Auth endpoint."""
try:
response = httpx.get(jwks_url, timeout=5.0)
response.raise_for_status()
return response.json()
except httpx.HTTPError as e:
print(f"❌ Failed to fetch JWKS: {e}")
raise
def get_signing_key(token: str, jwks_data: Dict[str, Any]) -> Dict[str, Any]:
"""Extract the public key from JWKS that matches the token's kid."""
try:
unverified_header = jwt.get_unverified_header(token)
kid = unverified_header.get("kid")
if not kid:
raise ValueError("Token missing key ID (kid)")
print(f"🔑 Token Key ID (kid): {kid}")
for key in jwks_data.get("keys", []):
if key.get("kid") == kid:
print(f"✅ Found matching public key in JWKS")
return key
raise ValueError(f"No matching key found for kid: {kid}")
except JWTError as e:
print(f"❌ Error extracting key from token: {e}")
raise
def verify_token(token: str, jwks_url: str, audience: str = None, issuer: str = None) -> Dict[str, Any]:
"""
Verify JWT token using JWKS.
Args:
token: JWT token string
jwks_url: URL to JWKS endpoint
audience: Expected audience (optional)
issuer: Expected issuer (optional)
Returns:
Decoded token payload
"""
print(f"🔍 Verifying JWT token...")
print(f"📍 JWKS URL: {jwks_url}")
# Fetch JWKS
jwks_data = get_jwks(jwks_url)
print(f"✅ JWKS fetched successfully")
# Get signing key
signing_key = get_signing_key(token, jwks_data)
# Verify token
try:
options = {
"verify_signature": True,
"verify_exp": True,
"verify_aud": bool(audience),
"verify_iss": bool(issuer),
}
payload = jwt.decode(
token,
signing_key,
algorithms=["EdDSA"],
audience=audience,
issuer=issuer,
options=options
)
print(f"✅ Token signature verified")
print(f"✅ Token is valid")
return payload
except jwt.ExpiredSignatureError:
print(f"❌ Token has expired")
raise
except jwt.JWTClaimsError as e:
print(f"❌ Invalid token claims: {e}")
raise
except JWTError as e:
print(f"❌ JWT verification failed: {e}")
raise
def main():
parser = argparse.ArgumentParser(description="Test JWT token verification with JWKS")
parser.add_argument("--jwks-url", required=True, help="JWKS endpoint URL")
parser.add_argument("--token", required=True, help="JWT token to verify")
parser.add_argument("--audience", help="Expected audience (optional)")
parser.add_argument("--issuer", help="Expected issuer (optional)")
args = parser.parse_args()
try:
payload = verify_token(
token=args.token,
jwks_url=args.jwks_url,
audience=args.audience,
issuer=args.issuer
)
print("\n" + "="*60)
print("✅ JWT Verification Successful!")
print("="*60)
print("\n📋 Token Payload:")
print(json.dumps(payload, indent=2))
print("\n🔐 User Information:")
print(f" - User ID (sub): {payload.get('sub')}")
print(f" - Email: {payload.get('email')}")
print(f" - Name: {payload.get('name')}")
print(f" - Issued At (iat): {payload.get('iat')}")
print(f" - Expires At (exp): {payload.get('exp')}")
except Exception as e:
print("\n" + "="*60)
print("❌ JWT Verification Failed")
print("="*60)
print(f"\nError: {e}")
return 1
return 0
if __name__ == "__main__":
exit(main())
#!/usr/bin/env python3
"""
Verify JWKS endpoint availability and structure.
Usage:
python verify_jwks.py <jwks-url>
python verify_jwks.py http://localhost:3000/api/auth/jwks
"""
import sys
import json
import httpx
from typing import Dict, Any
def verify_jwks(jwks_url: str) -> Dict[str, Any]:
"""
Fetch and verify JWKS endpoint.
Args:
jwks_url: URL to Better Auth JWKS endpoint
Returns:
JWKS data if successful
Raises:
Exception if JWKS endpoint is not accessible or invalid
"""
print(f"🔍 Fetching JWKS from: {jwks_url}")
try:
response = httpx.get(jwks_url, timeout=5.0)
response.raise_for_status()
jwks_data = response.json()
# Verify JWKS structure
if "keys" not in jwks_data:
raise ValueError("JWKS response missing 'keys' field")
keys = jwks_data["keys"]
if not isinstance(keys, list):
raise ValueError("JWKS 'keys' field must be a list")
if len(keys) == 0:
raise ValueError("JWKS 'keys' list is empty")
print(f"✅ JWKS endpoint is accessible")
print(f"✅ Found {len(keys)} public key(s)")
# Display each key
for i, key in enumerate(keys, 1):
print(f"\n📋 Public Key {i}:")
print(f" - Key ID (kid): {key.get('kid', 'N/A')}")
print(f" - Key Type (kty): {key.get('kty', 'N/A')}")
print(f" - Curve (crv): {key.get('crv', 'N/A')}")
print(f" - Public Key (x): {key.get('x', 'N/A')[:20]}...")
# Verify required fields
required_fields = ["kid", "kty", "crv", "x"]
missing = [f for f in required_fields if f not in key]
if missing:
print(f" ⚠️ Missing fields: {', '.join(missing)}")
print(f"\n✅ JWKS structure is valid")
return jwks_data
except httpx.HTTPError as e:
print(f"❌ HTTP Error: {e}")
print(f" Make sure Better Auth is running and JWT plugin is enabled")
raise
except json.JSONDecodeError as e:
print(f"❌ Invalid JSON response: {e}")
raise
except Exception as e:
print(f"❌ Verification failed: {e}")
raise
def main():
if len(sys.argv) < 2:
print("Usage: python verify_jwks.py <jwks-url>")
print("Example: python verify_jwks.py http://localhost:3000/api/auth/jwks")
sys.exit(1)
jwks_url = sys.argv[1]
try:
jwks_data = verify_jwks(jwks_url)
print("\n" + "="*60)
print("✅ JWKS Verification Complete!")
print("="*60)
print("\nNext steps:")
print("1. Copy asset templates to your FastAPI project")
print("2. Configure BETTER_AUTH_URL in backend .env")
print("3. Test JWT verification with test_jwt_verification.py")
except Exception:
print("\n" + "="*60)
print("❌ JWKS Verification Failed")
print("="*60)
sys.exit(1)
if __name__ == "__main__":
main()
Related skills
FAQ
How does the backend verify tokens?
The FastAPI backend fetches the JWKS from the Better Auth /api/auth/jwks endpoint and verifies the JWT signature against it.
Does Better Auth provide a useSession hook?
No. The skill notes Better Auth does not provide a useSession() hook and shows using authClient.getSession() inside useEffect.