I can register and login (neeed more tests)

This commit is contained in:
2025-10-26 17:00:00 +01:00
parent e12e4389b6
commit f39205dba0
16 changed files with 1363 additions and 3 deletions
+20
View File
@@ -1,9 +1,19 @@
annotated-doc==0.0.3
annotated-types==0.7.0
anyio==4.11.0 anyio==4.11.0
apsw==3.50.4.0 apsw==3.50.4.0
apswutils==0.1.0 apswutils==0.1.0
argon2-cffi==25.1.0
argon2-cffi-bindings==25.1.0
beautifulsoup4==4.14.2 beautifulsoup4==4.14.2
certifi==2025.10.5 certifi==2025.10.5
cffi==2.0.0
click==8.3.0 click==8.3.0
cryptography==46.0.3
dnspython==2.8.0
ecdsa==0.19.1
email-validator==2.3.0
fastapi==0.120.0
fastcore==1.8.13 fastcore==1.8.13
fastlite==0.2.1 fastlite==0.2.1
h11==0.16.0 h11==0.16.0
@@ -13,20 +23,30 @@ httpx==0.28.1
idna==3.11 idna==3.11
iniconfig==2.3.0 iniconfig==2.3.0
itsdangerous==2.2.0 itsdangerous==2.2.0
-e git+ssh://git@sheerka.synology.me:1010/kodjo/MyAuth.git@0138ac247a4a53dc555b94ec13119eba16e1db68#egg=myauth
oauthlib==3.3.1 oauthlib==3.3.1
packaging==25.0 packaging==25.0
passlib==1.7.4
pluggy==1.6.0 pluggy==1.6.0
pyasn1==0.6.1
pycparser==2.23
pydantic==2.12.3
pydantic-settings==2.11.0
pydantic_core==2.41.4
Pygments==2.19.2 Pygments==2.19.2
pytest==8.4.2 pytest==8.4.2
python-dateutil==2.9.0.post0 python-dateutil==2.9.0.post0
python-dotenv==1.1.1 python-dotenv==1.1.1
python-fasthtml==0.12.30 python-fasthtml==0.12.30
python-jose==3.5.0
python-multipart==0.0.20 python-multipart==0.0.20
PyYAML==6.0.3 PyYAML==6.0.3
rsa==4.9.1
six==1.17.0 six==1.17.0
sniffio==1.3.1 sniffio==1.3.1
soupsieve==2.8 soupsieve==2.8
starlette==0.48.0 starlette==0.48.0
typing-inspection==0.4.2
typing_extensions==4.15.0 typing_extensions==4.15.0
uvicorn==0.38.0 uvicorn==0.38.0
uvloop==0.22.1 uvloop==0.22.1
BIN
View File
Binary file not shown.
+31
View File
@@ -0,0 +1,31 @@
from fasthtml import serve
from fasthtml.components import Link, Script
from fasthtml.fastapp import fast_app
from myauth import create_auth_app_for_sqlite
from myfasthtml.auth.routes import setup_auth_routes
from myfasthtml.auth.utils import create_auth_beforeware
beforeware = create_auth_beforeware()
# Create FastHTML app
daisyui_offline_links = [
Link(href="./myfasthtml/assets/daisyui-5.css", rel="stylesheet", type="text/css"),
Link(href="./myfasthtml/assets/daisyui-5-themes.css", rel="stylesheet", type="text/css"),
Script(src="./myfasthtml/assets/tailwindcss-browser@4.js"),
]
daisyui_online_links = [
Link(rel='stylesheet', href='https://cdn.jsdelivr.net/npm/daisyui@5/daisyui.css'),
Script(src='https://cdn.jsdelivr.net/npm/@tailwindcss/browser@4'),
]
app, rt = fast_app(
# before=beforeware,
hdrs=tuple(daisyui_offline_links)
)
# Setup authentication routes
setup_auth_routes(app, rt)
if __name__ == "__main__":
serve()
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
+283
View File
@@ -0,0 +1,283 @@
# FastHTML + FastAPI Authentication Integration
This package provides a complete authentication system integrating FastHTML (frontend) with FastAPI (backend).
## Architecture
```
auth/
├── __init__.py # Package initialization
├── utils.py # JWT helpers + Beforeware
├── pages/
│ ├── __init__.py
│ ├── LoginPage.py # Login form component
│ ├── RegisterPage.py # Registration form component
│ └── WelcomePage.py # Protected welcome page
└── routes.py # All routes (public + protected)
```
## Features
- ✅ **Login page** with email/password authentication
- ✅ **Registration page** with password confirmation
- ✅ **Automatic token refresh** (proactive refresh at 5 minutes before expiry)
- ✅ **Route protection** with Beforeware (whitelist support)
- ✅ **Session management** using FastHTML signed cookies
- ✅ **DaisyUI 5** components for beautiful UI
- ✅ **HTMX integration** for dynamic form submissions
## Dependencies
Install the required packages:
```bash
pip install fasthtml httpx python-jose[cryptography] --break-system-packages
```
Note: `python-jose` may already be installed if you have FastAPI.
## Configuration
### 1. Update API configuration in `auth/utils.py`
```python
API_BASE_URL = "http://localhost:5001" # Your FastAPI backend URL
JWT_SECRET = "jwt-secret-to-change" # Must match your FastAPI secret
```
### 2. Create your `main.py`
```python
from fasthtml.common import fast_app, serve, Link, Script
from auth import create_auth_beforeware, setup_auth_routes
# Import your existing FastAPI auth app
# from your_auth_module import create_auth_app_for_sqlite
# Create Beforeware for route protection
beforeware = create_auth_beforeware()
# Create FastHTML app
app, rt = fast_app(
before=beforeware,
hdrs=(
# DaisyUI 5 CSS
Link(rel='stylesheet', href='https://cdn.jsdelivr.net/npm/daisyui@5/daisyui.css'),
# Tailwind CSS 4
Script(src='https://cdn.jsdelivr.net/npm/@tailwindcss/browser@4'),
)
)
# Mount FastAPI auth backend
# auth_api = create_auth_app_for_sqlite("Users.db", "jwt-secret-to-change")
# app.mount("/auth", auth_api)
# Setup authentication routes
setup_auth_routes(app, rt)
if __name__ == "__main__":
serve()
```
## Usage
### Starting the application
```bash
python main.py
```
The application will be available at `http://localhost:5001`.
### Routes
**Public routes:**
- `GET /login` - Display login page
- `POST /login` - Handle login submission
- `GET /register` - Display registration page
- `POST /register` - Handle registration submission
**Protected routes (require authentication):**
- `GET /` - Welcome page (home)
- `GET /welcome` - Alternative welcome page
- `POST /logout` - Logout and revoke tokens
### Adding custom protected routes
```python
@rt("/dashboard")
def get_dashboard(session, auth):
"""
Custom protected route.
The 'auth' parameter is automatically injected by the Beforeware.
"""
user_info = session.get('user_info', {})
return Div(
H1(f"Dashboard for {user_info.get('email', 'User')}"),
# Your dashboard content
)
```
### Extending the whitelist
If you need to add more public routes:
```python
beforeware = create_auth_beforeware(
additional_patterns=[
'/about',
'/contact',
r'/api/.*', # Regex patterns are supported
]
)
```
## How it works
### 1. Beforeware Protection
The `auth_before()` function runs before every route (except whitelisted ones):
1. Checks if `access_token` exists in session
2. Validates token expiration
3. If token expires in < 5 minutes, automatically refreshes it
4. If refresh fails or no token exists, redirects to `/login`
### 2. Token Refresh Flow
```
User Request → Beforeware checks token
Token expires in < 5 min?
↓ Yes
Call /auth/refresh with refresh_token
Update session with new tokens
Continue to route handler
```
### 3. Session Storage
Session contains:
- `access_token` (JWT, 30 minutes validity)
- `refresh_token` (Opaque, 7 days validity)
- `user_info` (email, id, etc.)
## Customization
### Changing the refresh threshold
Edit `auth/utils.py`:
```python
TOKEN_REFRESH_THRESHOLD_MINUTES = 5 # Change to your preferred value
```
### Customizing page components
Each page component is a class with:
- `__init__()` - Constructor with parameters
- `_render()` - HTML generation logic
- `__ft__()` - FastHTML protocol method
Example:
```python
class LoginPage:
def __init__(self, error_message=None, redirect_url="/"):
self.error_message = error_message
self.redirect_url = redirect_url
def _render(self):
# Build your custom HTML here
return Div(...)
def __ft__(self):
return self._render()
```
### Styling with DaisyUI 5
DaisyUI 5 provides semantic class names:
```python
# Button
Button("Click me", cls="btn btn-primary")
# Input
Input(type="email", cls="input input-bordered")
# Card
Div(
Div(
H2("Title", cls="card-title"),
P("Content"),
cls="card-body"
),
cls="card bg-base-100 shadow-xl"
)
# Alert
Div(
Span("Success message"),
cls="alert alert-success"
)
```
## Troubleshooting
### Issue: "Module not found: auth"
Make sure the `auth` directory is in your Python path or in the same directory as `main.py`.
### Issue: "Connection refused" when calling API
Check that:
1. `API_BASE_URL` in `auth/utils.py` matches your FastAPI server
2. Your FastAPI backend is running
3. The `/auth` endpoints are correctly mounted
### Issue: Token refresh not working
Verify:
1. `JWT_SECRET` in `auth/utils.py` matches your FastAPI secret
2. Your FastAPI `/auth/refresh` endpoint is working
3. The refresh token is being stored correctly in the session
### Issue: Infinite redirect loop
Check:
1. The `/login` route is in the whitelist (it should be by default)
2. Your FastAPI `/auth/login` endpoint returns `access_token` and `refresh_token`
## Testing
To test the authentication flow:
1. Start your application
2. Navigate to `http://localhost:5001/`
3. You should be redirected to `/login`
4. Register a new account at `/register`
5. Login with your credentials
6. You should see the welcome page
7. Your session will automatically refresh tokens as needed
## API Integration
This package expects your FastAPI backend to provide these endpoints:
- `POST /auth/login` - Returns `access_token` and `refresh_token`
- `POST /auth/register` - Creates a new user
- `POST /auth/refresh` - Refreshes the access token
- `GET /auth/me` - Returns current user info (requires Bearer token)
- `POST /auth/logout` - Revokes refresh token
## License
This code is provided as-is for educational and development purposes.
## Questions?
Feel free to modify and extend this authentication system according to your needs!
View File
@@ -2,8 +2,7 @@ from fasthtml.components import *
class LoginPage: class LoginPage:
def __init__(self, settings_manager, error_message=None, success_message=None): def __init__(self, error_message=None, success_message=None):
self.settings_manager = settings_manager
self.error_message = error_message self.error_message = error_message
self.success_message = success_message self.success_message = success_message
@@ -66,7 +65,7 @@ class LoginPage:
cls="btn w-full font-bold py-2 px-4 rounded" cls="btn w-full font-bold py-2 px-4 rounded"
), ),
action=ROUTE_ROOT + Routes.LoginByEmail, action="/login-p",
method="post", method="post",
cls="mb-6" cls="mb-6"
), ),
@@ -84,3 +83,6 @@ class LoginPage:
) )
) )
def __ft__(self):
return self.render()
+116
View File
@@ -0,0 +1,116 @@
from fasthtml.components import *
class RegisterPage:
def __init__(self, error_message: str = None):
self.error_message = error_message
def register_page(self, error_message: str):
self.error_message = error_message
return self.__ft__()
def __ft__(self):
"""
Create the registration page.
Args:
error_message: Optional error message to display
Returns:
Components representing the registration page
"""
# Create alert for error message
error_alert = None
if self.error_message:
error_alert = Div(
P(self.error_message, cls="text-sm"),
cls="bg-soft bg-error border border-red-400 text-red-700 px-4 py-3 rounded mb-4"
)
return Div(
# Page title
H1("Create an Account", cls="text-3xl font-bold text-center mb-6"),
# Registration Form
Div(
# Error alert
error_alert if error_alert else "",
Form(
# Username field
Div(
Label("Username", For="username", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="text",
id="username",
name="username",
placeholder="Choose a username",
required=True,
minlength=3,
maxlength=30,
pattern="[a-zA-Z0-9_-]+",
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
P("Only letters, numbers, underscores, and hyphens", cls="text-xs text-gray-500 mt-1"),
cls="mb-4"
),
# Email field
Div(
Label("Email", For="email", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="email",
id="email",
name="email",
placeholder="you@example.com",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-4"
),
# Password field
Div(
Label("Password", For="password", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="password",
id="password",
name="password",
placeholder="Create a password",
required=True,
minlength=8,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
P("At least 8 characters with uppercase, lowercase, and number", cls="text-xs text-gray-500 mt-1"),
cls="mb-4"
),
# Confirm password field
Div(
Label("Confirm Password", For="confirm_password", cls="block text-sm font-medium text-gray-700 mb-1"),
Input(
type="password",
id="confirm_password",
name="confirm_password",
placeholder="Confirm your password",
required=True,
cls="w-full px-3 py-2 border rounded-md focus:outline-none focus:ring-2"
),
cls="mb-6"
),
# Submit button
Button(
"Create Account",
type="submit",
cls="btn w-full font-bold py-2 px-4 rounded"
),
action="register-p",
method="post",
cls="mb-6"
),
cls="p-8 rounded-lg shadow-2xl max-w-md mx-auto"
)
)
@@ -0,0 +1,216 @@
"""
Register page component for FastHTML application.
Provides a styled registration form using DaisyUI 5 components.
"""
from fasthtml.common import Div, Form, Input, Button, H1, P, A, Label
class RegisterPage:
"""
Register page component with email and password fields.
Attributes:
error_message: Optional error message to display
redirect_url: URL to redirect after successful registration
"""
def __init__(self, error_message: str = None, redirect_url: str = "/login"):
"""
Initialize register page.
Args:
error_message: Optional error message to display
redirect_url: URL to redirect after successful registration (default: "/login")
"""
self.error_message = error_message
self.redirect_url = redirect_url
def _render(self):
"""
Render the register page HTML structure.
Returns:
FastHTML component tree for the register page
"""
# Error alert component (only shown if error_message exists)
error_alert = None
if self.error_message:
error_alert = Div(
Div(
self._svg_icon(),
self.error_message,
cls="flex items-center gap-2"
),
role="alert",
cls="alert alert-error mb-4",
id="error-alert"
)
# Success message container (empty by default, filled by HTMX on success)
success_alert = Div(id="success-alert")
# Register form
register_form = Form(
# Hidden field for redirect URL
Input(type="hidden", name="redirect_url", value=self.redirect_url),
# Email input
Label(
Div("Email", cls="label-text"),
cls="form-control w-full mb-4"
)(
Input(
type="email",
name="email",
placeholder="[email protected]",
required=True,
cls="input input-bordered w-full"
)
),
# Password input
Label(
Div("Password", cls="label-text"),
Div("Must be at least 8 characters", cls="label-text-alt text-base-content/60"),
cls="form-control w-full mb-4"
)(
Input(
type="password",
name="password",
placeholder="Create a password",
required=True,
minlength="8",
cls="input input-bordered w-full"
)
),
# Confirm password input
Label(
Div("Confirm Password", cls="label-text"),
cls="form-control w-full mb-4"
)(
Input(
type="password",
name="confirm_password",
placeholder="Confirm your password",
required=True,
minlength="8",
cls="input input-bordered w-full"
)
),
# Submit button
Button(
"Create Account",
type="submit",
cls="btn btn-primary w-full"
),
method="post",
action="/register",
hx_post="/register",
hx_target="#register-card",
hx_swap="outerHTML"
)
# Main card container
card = Div(
Div(
# Card title
H1("Create Account", cls="text-3xl font-bold text-center mb-2"),
P("Sign up for a new account", cls="text-center text-base-content/70 mb-6"),
# Success alert
success_alert,
# Error alert
error_alert if error_alert else Div(),
# Register form
register_form,
# Login link
Div(
P(
"Already have an account? ",
A("Sign in", href="/login", cls="link link-primary"),
cls="text-center text-sm"
),
cls="mt-4"
),
cls="card-body"
),
cls="card bg-base-100 w-96 shadow-xl",
id="register-card"
)
# Full page container
return Div(
card,
cls="flex items-center justify-center min-h-screen bg-base-200"
)
def _svg_icon(self):
"""
Helper method to create error icon SVG.
Returns:
SVG icon as NotStr
"""
from fasthtml.common import NotStr
return NotStr('''
<svg xmlns="http://www.w3.org/2000/svg"
class="h-6 w-6 shrink-0 stroke-current"
fill="none"
viewBox="0 0 24 24">
<path stroke-linecap="round"
stroke-linejoin="round"
stroke-width="2"
d="M10 14l2-2m0 0l2-2m-2 2l-2-2m2 2l2 2m7-2a9 9 0 11-18 0 9 9 0 0118 0z" />
</svg>
''')
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered register page component
"""
return self._render()
def create_success_message(message: str = "Account created successfully!"):
"""
Create a success alert component for HTMX response.
Args:
message: Success message to display
Returns:
Success alert component
"""
from fasthtml.common import NotStr
return Div(
Div(
NotStr('''
<svg xmlns="http://www.w3.org/2000/svg"
class="h-6 w-6 shrink-0 stroke-current"
fill="none"
viewBox="0 0 24 24">
<path stroke-linecap="round"
stroke-linejoin="round"
stroke-width="2"
d="M9 12l2 2 4-4m6 2a9 9 0 11-18 0 9 9 0 0118 0z" />
</svg>
'''),
message,
cls="flex items-center gap-2"
),
role="alert",
cls="alert alert-success mb-4"
)
+185
View File
@@ -0,0 +1,185 @@
"""
Welcome page component for FastHTML application.
Provides a protected welcome page that displays user information.
This page is only accessible to authenticated users.
"""
from fasthtml.common import Div, H1, H2, P, Button, A
from typing import Optional, Dict, Any
class WelcomePage:
"""
Welcome page component for authenticated users.
Attributes:
user_info: Dictionary containing user information (email, id, etc.)
redirect_url: URL for logout redirect
"""
def __init__(self, user_info: Optional[Dict[str, Any]] = None, redirect_url: str = "/login"):
"""
Initialize welcome page.
Args:
user_info: Dictionary containing user information
redirect_url: URL to redirect after logout (default: "/login")
"""
self.user_info = user_info or {}
self.redirect_url = redirect_url
def _render(self):
"""
Render the welcome page HTML structure.
Returns:
FastHTML component tree for the welcome page
"""
# Extract user info
email = self.user_info.get('email', 'Guest')
user_id = self.user_info.get('id', 'N/A')
# Hero section
hero = Div(
Div(
# Welcome message
H1(
f"Welcome back, {email}!",
cls="text-5xl font-bold"
),
P(
"You have successfully logged in to your account.",
cls="py-6 text-lg"
),
# Action buttons
Div(
Button(
"Logout",
cls="btn btn-primary",
hx_post="/logout",
hx_target="body",
hx_swap="outerHTML"
),
cls="flex gap-2"
),
cls="max-w-md"
),
cls="hero min-h-screen bg-base-200"
)
# User info card
info_card = Div(
Div(
H2("Your Information", cls="card-title mb-4"),
# User details
Div(
self._info_row("Email", email),
self._info_row("User ID", str(user_id)),
cls="space-y-2"
),
cls="card-body"
),
cls="card bg-base-100 shadow-xl w-full max-w-md"
)
# Main container
return Div(
# Hero section
hero,
# Info card (positioned absolutely over hero)
Div(
info_card,
cls="absolute top-20 right-20 hidden lg:block"
),
cls="relative"
)
def _info_row(self, label: str, value: str):
"""
Create an information row for displaying user details.
Args:
label: Label for the information
value: Value to display
Returns:
Div containing label and value
"""
return Div(
Div(
P(label, cls="font-semibold text-sm text-base-content/70"),
P(value, cls="text-base"),
cls="flex flex-col"
),
cls="py-2 border-b border-base-300 last:border-b-0"
)
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered welcome page component
"""
return self._render()
class SimpleWelcomePage:
"""
Simpler version of welcome page without user info card.
Useful for quick prototyping or minimal designs.
"""
def __init__(self, user_email: str = "Guest"):
"""
Initialize simple welcome page.
Args:
user_email: User's email address
"""
self.user_email = user_email
def _render(self):
"""
Render the simple welcome page.
Returns:
FastHTML component tree
"""
return Div(
Div(
Div(
H1(f"Welcome, {self.user_email}!", cls="text-4xl font-bold mb-4"),
P("You are now logged in.", cls="mb-6 text-lg"),
Button(
"Logout",
cls="btn btn-primary",
hx_post="/logout",
hx_target="body",
hx_swap="outerHTML"
),
cls="text-center"
),
cls="hero-content"
),
cls="hero min-h-screen bg-base-200"
)
def __ft__(self):
"""
FastHTML rendering protocol method.
Returns:
Rendered simple welcome page
"""
return self._render()
+194
View File
@@ -0,0 +1,194 @@
"""
Authentication routes for FastHTML application.
Provides routes for login, register, logout, and protected pages.
"""
from fasthtml.common import RedirectResponse
from myauth import create_auth_app_for_sqlite
from ..auth.pages.LoginPage import LoginPage
from ..auth.pages.RegisterPage import RegisterPage
from ..auth.pages.WelcomePage import WelcomePage
from ..auth.utils import (
login_user,
register_user,
logout_user,
get_user_info
)
def setup_auth_routes(app, rt, mount_auth_app=True):
"""
Setup all authentication and protected routes.
Args:
app: FastHTML application instance
rt: Route decorator from FastHTML
mount_auth_app: Whether to mount the auth FastApi API routes
"""
# ============================================================================
# PUBLIC ROUTES (Login & Register)
# ============================================================================
@rt("/login")
def get(error: str = None):
"""
Display login page.
Args:
error: Optional error message from query params
Returns:
LoginPage component
"""
return LoginPage(error_message=error)
@rt("/login-p")
def post(email: str, password: str, session, redirect_url: str = "/"):
"""
Handle login form submission.
Args:
email: User email from form
password: User password from form
session: FastHTML session object
redirect_url: URL to redirect after successful login
Returns:
RedirectResponse on success, or LoginPage with error on failure
"""
# Attempt login
auth_data = login_user(email, password)
if auth_data:
# Login successful - store tokens in session
session['access_token'] = auth_data['access_token']
session['refresh_token'] = auth_data['refresh_token']
# Get user info and store in session
user_info = get_user_info(auth_data['access_token'])
if user_info:
session['user_info'] = user_info
# Redirect to protected page
return RedirectResponse(redirect_url, status_code=303)
else:
# Login failed - return error message via HTMX
return LoginPage(error_message="Invalid email or password. Please try again.")
@rt("/register")
def get(error: str = None):
"""
Display registration page.
Args:
error: Optional error message from query params
Returns:
RegisterPage component
"""
return RegisterPage(error_message=error)
@rt("/register-p")
def post(email: str, username: str, password: str, confirm_password: str, session):
"""
Handle registration form submission.
Args:
email: User email from form
password: User password from form
confirm_password: Password confirmation from form
session: FastHTML session object
Returns:
RegisterPage with success/error message via HTMX
"""
# Validate password confirmation
# if password != confirm_password:
# return RegisterPage(error_message="Passwords do not match. Please try again.")
#
# # Validate password length
# if len(password) < 8:
# return RegisterPage(error_message="Password must be at least 8 characters long.")
# Attempt registration
result = register_user(email, username, password)
if result:
# Registration successful - show success message and auto-login
auth_data = login_user(email, password)
if auth_data:
# Store tokens in session
session['access_token'] = auth_data['access_token']
session['refresh_token'] = auth_data['refresh_token']
# Get user info and store in session
user_info = get_user_info(auth_data['access_token'])
if user_info:
session['user_info'] = user_info
# Redirect to welcome page
return RedirectResponse("/", status_code=303)
else:
# Auto-login failed, redirect to login page
return RedirectResponse("/login", status_code=303)
else:
# Registration failed
return RegisterPage(error_message="Registration failed. Email may already be in use.")
# ============================================================================
# PROTECTED ROUTES (Require authentication)
# ============================================================================
@rt("/welcome")
def get(session, auth):
"""
Alternative welcome page route (protected).
Args:
session: FastHTML session object
auth: User auth info from request scope
Returns:
WelcomePage component
"""
user_info = session.get('user_info', {})
return WelcomePage(user_info=user_info)
@rt("/logout")
def post(session):
"""
Handle logout request.
Revokes refresh token and clears session.
Args:
session: FastHTML session object
Returns:
RedirectResponse to login page
"""
# Get refresh token from session
refresh_token = session.get('refresh_token')
# Revoke refresh token on backend
if refresh_token:
logout_user(refresh_token)
# Clear session
session.clear()
# Redirect to login page
return RedirectResponse("/login", status_code=303)
def mount_auth_fastapi_api():
# Mount FastAPI auth backend
auth_api = create_auth_app_for_sqlite("Users.db", "jwt-secret-to-change")
app.mount("/auth", auth_api)
if mount_auth_app:
mount_auth_fastapi_api()
+297
View File
@@ -0,0 +1,297 @@
"""
Authentication utilities for FastHTML application.
This module provides:
- JWT token validation and refresh logic
- Beforeware for protecting routes
- Helper functions for API communication
"""
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List
import httpx
from fasthtml.common import RedirectResponse, Beforeware
from jose import jwt, JWTError
# Configuration
API_BASE_URL = "http://localhost:5001" # Base URL for FastAPI backend
JWT_SECRET = "jwt-secret-to-change" # Must match FastAPI secret
JWT_ALGORITHM = "HS256"
TOKEN_REFRESH_THRESHOLD_MINUTES = 5 # Refresh token if expires in less than 5 minutes
# Default patterns to skip authentication
DEFAULT_SKIP_PATTERNS = [
r'/favicon\.ico',
r'/static/.*',
r'.*\.css',
r'.*\.js',
'/login',
'/login2',
'/register',
]
def create_auth_beforeware(additional_patterns: Optional[List[str]] = None) -> Beforeware:
"""
Create a Beforeware instance for route protection.
Args:
additional_patterns: Optional list of additional URL patterns to skip authentication
Returns:
Beforeware instance configured for authentication
"""
patterns = DEFAULT_SKIP_PATTERNS.copy()
if additional_patterns:
patterns.extend(additional_patterns)
return Beforeware(auth_before, skip=patterns)
def auth_before(request, session):
"""
Beforeware function that runs before each protected route.
Checks authentication status and automatically refreshes tokens if needed.
Redirects to login if authentication fails.
Args:
request: Starlette request object
session: FastHTML session object
Returns:
RedirectResponse to login page if authentication fails, None otherwise
"""
# Get tokens from session
access_token = session.get('access_token')
refresh_token = session.get('refresh_token')
print(f"path={request.scope['path']}, {session=}, {access_token=}, {refresh_token=}")
# If no access token, redirect to login
if not access_token:
return RedirectResponse('/login', status_code=303)
# Validate access token and check expiration
try:
payload = decode_jwt(access_token)
exp_timestamp = payload.get('exp')
if exp_timestamp:
exp_datetime = datetime.fromtimestamp(exp_timestamp, tz=timezone.utc)
now = datetime.now(timezone.utc)
time_until_expiry = (exp_datetime - now).total_seconds() / 60
# If token expires in less than 5 minutes, try to refresh
if time_until_expiry < TOKEN_REFRESH_THRESHOLD_MINUTES:
if refresh_token:
new_tokens = refresh_access_token(refresh_token)
if new_tokens:
# Update session with new tokens
session['access_token'] = new_tokens['access_token']
session['refresh_token'] = new_tokens['refresh_token']
# Store auth info in request scope
request.scope['auth'] = session.get('user_info')
return None
else:
# Refresh failed, redirect to login
return RedirectResponse('/login', status_code=303)
else:
# No refresh token available, redirect to login
return RedirectResponse('/login', status_code=303)
# Token is valid, store auth info in request scope
request.scope['auth'] = session.get('user_info')
return None
except JWTError:
# Token is invalid or expired, try to refresh
if refresh_token:
new_tokens = refresh_access_token(refresh_token)
if new_tokens:
# Update session with new tokens
session['access_token'] = new_tokens['access_token']
session['refresh_token'] = new_tokens['refresh_token']
request.scope['auth'] = session.get('user_info')
return None
# Could not refresh, redirect to login
return RedirectResponse('/login', status_code=303)
def decode_jwt(token: str) -> Dict[str, Any]:
"""
Decode and validate a JWT token.
Args:
token: JWT token string
Returns:
Dictionary containing the token payload
Raises:
JWTError: If token is invalid or expired
"""
return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
def check_token_expiry(token: str) -> Optional[float]:
"""
Check how many minutes until token expires.
Args:
token: JWT token string
Returns:
Minutes until expiration, or None if token is invalid
"""
try:
payload = decode_jwt(token)
exp_timestamp = payload.get('exp')
if exp_timestamp:
exp_datetime = datetime.fromtimestamp(exp_timestamp, tz=timezone.utc)
now = datetime.now(timezone.utc)
time_until_expiry = (exp_datetime - now).total_seconds() / 60
return time_until_expiry
return None
except JWTError:
return None
def login_user(email: str, password: str) -> Optional[Dict[str, Any]]:
"""
Authenticate user with email and password.
Args:
email: User email address
password: User password
Returns:
Dictionary containing access_token, refresh_token, and user_info if successful,
None if authentication fails
"""
try:
response = httpx.post(
f"{API_BASE_URL}/auth/login",
data={"username": email, "password": password},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=10.0
)
if response.status_code == 200:
data = response.json()
return {
'access_token': data.get('access_token'),
'refresh_token': data.get('refresh_token'),
'token_type': data.get('token_type'),
}
return None
except httpx.HTTPError:
return None
def register_user(email: str, username: str, password: str) -> Optional[Dict[str, Any]]:
"""
Register a new user.
Args:
email: User email address
password: User password
Returns:
Dictionary containing success message if registration succeeds,
None if registration fails
"""
try:
response = httpx.post(
f"{API_BASE_URL}/auth/register",
json={"email": email, "username": username, "password": password},
timeout=10.0
)
if response.status_code == 200:
return response.json()
return None
except httpx.HTTPError:
return None
def refresh_access_token(refresh_token: str) -> Optional[Dict[str, Any]]:
"""
Refresh the access token using a refresh token.
Args:
refresh_token: Valid refresh token
Returns:
Dictionary containing new access_token and refresh_token if successful,
None if refresh fails
"""
try:
response = httpx.post(
f"{API_BASE_URL}/auth/refresh",
json={"refresh_token": refresh_token},
timeout=10.0
)
if response.status_code == 200:
data = response.json()
return {
'access_token': data.get('access_token'),
'refresh_token': data.get('refresh_token'),
}
return None
except httpx.HTTPError:
return None
def get_user_info(access_token: str) -> Optional[Dict[str, Any]]:
"""
Get current user information using access token.
Args:
access_token: Valid access token
Returns:
Dictionary containing user information if successful,
None if request fails
"""
try:
response = httpx.get(
f"{API_BASE_URL}/auth/me",
headers={"Authorization": f"Bearer {access_token}"},
timeout=10.0
)
if response.status_code == 200:
return response.json()
return None
except httpx.HTTPError:
return None
def logout_user(refresh_token: str) -> bool:
"""
Logout user by revoking the refresh token.
Args:
refresh_token: Refresh token to revoke
Returns:
True if logout successful, False otherwise
"""
try:
response = httpx.post(
f"{API_BASE_URL}/auth/logout",
json={"refresh_token": refresh_token},
timeout=10.0
)
return response.status_code == 200
except httpx.HTTPError:
return False