Flaxon 0.3.0 and Teloce-Py 0.3.0 are available on PyPI. Release instructions are maintained on this site; source links identify the framework snapshot. Check version and upgrade guidance.
04 | Customer authentication and sessions
Goal
Register, sign in, sign out and reject requests without valid sessions.
What you are building
validation.py checks JSON and field types before handlers write anything. security.py reads hashed session tokens, compares CSRF tokens, checks browser origins and rotates sessions after authentication changes. Password hashing runs in a worker thread because Argon2 is CPU intensive. The login handler verifies a dummy hash for unknown emails. The durable limiter counts attempts inside a transaction. Customer accounts are separate from staff Admin accounts. Start by requesting /api/auth/session; this issues an anonymous cookie and CSRF token. Registration and login replace both, so use the new values for the next request.
Build this chapter
Stop the server. Work in project_manager. Follow each CREATE, whole-file replacement, or marked-block edit below in order.
Step 1: CREATE - make parent folders, then create this file - validation.py
Validate JSON before passing values to ORM calls. Reuse these checks in all mutations.
Create the parent directories, then save this complete file.
"""Validate every write on the server, even when the browser already validated it."""
import re
from datetime import date
from flaxon.exceptions import BadRequest
async def json_object(request):
try:
data = await request.json()
except (ValueError, UnicodeDecodeError):
raise BadRequest("Send a valid JSON object.")
if not isinstance(data, dict):
raise BadRequest("Send a JSON object.")
return data
def text(data, key, maximum=120, required=True):
value = data.get(key, "")
if not isinstance(value, str):
raise BadRequest(f"{key} must be text.")
value = value.strip()
if (required and not value) or len(value) > maximum:
raise BadRequest(
f"{key} must contain 1 to {maximum} characters."
if required
else f"{key} must contain at most {maximum} characters."
)
return value
def credentials(data):
email = text(data, "email", 254).lower()
if not re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", email):
raise BadRequest("Enter a valid email address.")
password = data.get("password")
if not isinstance(password, str) or not 12 <= len(password) <= 128:
raise BadRequest("Use a password with 12 to 128 characters.")
return email, password
def task_fields(data, partial=False):
result = {}
if not partial or "title" in data:
result["title"] = text(data, "title", 160)
if not partial or "status" in data:
status = data.get("status", "todo")
if status not in ("todo", "doing", "done"):
raise BadRequest("Status must be todo, doing, or done.")
result["status"] = status
if not partial or "due_date" in data:
due = data.get("due_date") or None
if due is not None:
try:
due = date.fromisoformat(due).isoformat()
except (ValueError, TypeError):
raise BadRequest("Use YYYY-MM-DD for a due date.")
result["due_date"] = due
if not result:
raise BadRequest("Supply at least one editable field.")
return result
Say: "Validate JSON before passing values to ORM calls. Reuse these checks in all mutations."
Type: Follow the action above in validation.py; explain each new function or binding as you add it.
Viewers should see: The saved file matches this step. After all steps, run the chapter checks below and show the expected result.
Step 2: CREATE - make parent folders, then create this file - security.py
Read each helper in order: identify a session, require its user, check CSRF, rotate credentials, then limit attempts.
Create the parent directories, then save this complete file.
"""Opaque cookie sessions and session-bound CSRF; account ownership stays on the server."""
import asyncio
import hashlib
import hmac
import secrets
import time
from models import User, Session, AuthAttempt
from tortoise.expressions import F
from flaxon.db import in_transaction
from flaxon.exceptions import Forbidden, Unauthorized, TooManyRequests
from flaxon.http import JSONResponse
from flaxon.http.cookies import Cookie
COOKIE_NAME = "project_session"
SESSION_SECONDS = 8 * 60 * 60
def digest(value):
return hashlib.sha256(value.encode()).hexdigest()
async def read_session(request):
token = request.cookies.get(COOKIE_NAME, "")
if not token:
return None
return (
await Session.filter(token_hash=digest(token), expires_at__gt=int(time.time()))
.first()
.values()
)
async def require_user(request):
session = await read_session(request)
if not session or session["user_id"] is None:
raise Unauthorized("Please sign in.")
user = (
await User.filter(id=session["user_id"]).first().values("id", "name", "email")
)
if user is None:
raise Unauthorized("Please sign in.")
return user
async def check_csrf(request):
origin = request.headers.get("origin")
if origin and origin != request.app.public_origin:
raise Forbidden("This request came from an untrusted origin.")
if "application/json" not in request.headers.get("content-type", "").lower():
raise Forbidden("Send JSON with a CSRF token.")
session = await read_session(request)
supplied = request.headers.get("x-csrf-token", "")
if (
not session
or not supplied
or not hmac.compare_digest(session["csrf"], supplied)
):
raise Forbidden("Your session expired. Reload the page and try again.")
async def session_response(request, user=None, status=200):
old = request.cookies.get(COOKIE_NAME, "")
token, csrf = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
now = int(time.time())
async with in_transaction():
if old:
await Session.filter(token_hash=digest(old)).delete()
await Session.filter(expires_at__lte=now).delete()
await Session.create(
token_hash=digest(token),
user_id=user["id"] if user else None,
csrf=csrf,
expires_at=now + SESSION_SECONDS,
)
cookie = Cookie(
COOKIE_NAME,
token,
path="/",
httponly=True,
samesite="Lax",
secure=not request.app.debug,
max_age=SESSION_SECONDS,
)
response = JSONResponse(
{"data": {"user": user, "csrf": csrf}},
status_code=status,
headers={"cache-control": "no-store"},
)
response.headers.add("set-cookie", cookie.to_header())
return response
async def limit_auth_attempts(request, email):
"""A durable, single-instance limiter; production proxies also limit the auth routes."""
peer = request.scope.get("client") or ("unknown", 0)
key = digest(str(peer[0]) + ":" + email)
now = int(time.time())
async with in_transaction():
await AuthAttempt.filter(started_at__lt=now - 900).delete()
await AuthAttempt.get_or_create(key=key, defaults={"started_at": now})
await AuthAttempt.filter(key=key).update(attempts=F("attempts") + 1)
record = await AuthAttempt.get(key=key)
attempts = record.attempts
if attempts > 10:
raise TooManyRequests("Too many attempts. Try again in 15 minutes.")
Say: "Read each helper in order: identify a session, require its user, check CSRF, rotate credentials, then limit attempts."
Type: Follow the action above in security.py; explain each new function or binding as you add it.
Viewers should see: The saved file matches this step. After all steps, run the chapter checks below and show the expected result.
Step 3: CREATE - make parent folders, then create this file - modules/auth/init.py
This file owns the feature named by its module or component. Read the route/form flow before continuing.
Create the parent directories, then save this complete file.
"""Module-owned APIs and interface files."""
Say: "This file owns the feature named by its module or component. Read the route/form flow before continuing."
Type: Follow the action above in modules/auth/__init__.py; explain each new function or binding as you add it.
Viewers should see: The saved file matches this step. After all steps, run the chapter checks below and show the expected result.
Step 4: CREATE - make parent folders, then create this file - modules/auth/module.py
Create the auth module with session, registration, login and logout routes. Start with the anonymous CSRF handshake; registration rotates the cookie and token.
Create the parent directories, then save this complete file.
"""Registration and cookie authentication, separate from staff Admin accounts."""
import asyncio
from tortoise.exceptions import IntegrityError
from models import User
from pathlib import Path
from argon2.exceptions import VerifyMismatchError, VerificationError
from flaxon.modules import FlaxonModule
from flaxon.exceptions import Conflict, Unauthorized
from flaxon.http import JSONResponse
from security import (
check_csrf,
read_session,
require_user,
session_response,
limit_auth_attempts,
)
from validation import json_object, text, credentials
auth = FlaxonModule(
"auth", ui_dir=Path(__file__).parent / "ui", ui_routes={"Login.js": "/login"}
)
@auth.get("/session")
async def session(request):
current = await read_session(request)
if current:
user = await require_user(request) if current["user_id"] else None
return JSONResponse(
{"data": {"user": user, "csrf": current["csrf"]}},
headers={"cache-control": "no-store"},
)
return await session_response(request)
@auth.post("/register")
async def register(request):
await check_csrf(request)
data = await json_object(request)
email, password = credentials(data)
name = text(data, "name", 80)
await limit_auth_attempts(request, email)
password_hash = await asyncio.to_thread(request.app.hasher.hash, password)
try:
user = await User.create(name=name, email=email, password_hash=password_hash)
user_id = user.id
except IntegrityError:
raise Conflict("Unable to create this account. Try signing in.")
return await session_response(
request, {"id": user_id, "name": name, "email": email}, status=201
)
@auth.post("/login")
async def login(request):
await check_csrf(request)
email, password = credentials(await json_object(request))
await limit_auth_attempts(request, email)
user = await User.filter(email=email).first().values()
password_hash = user["password_hash"] if user else request.app.dummy_password_hash
try:
await asyncio.to_thread(request.app.hasher.verify, password_hash, password)
except (VerifyMismatchError, VerificationError):
raise Unauthorized("Email or password is incorrect.")
if user is None:
raise Unauthorized("Email or password is incorrect.")
public_user = {key: user[key] for key in ("id", "name", "email")}
return await session_response(request, public_user)
@auth.post("/logout")
async def logout(request):
await check_csrf(request)
return await session_response(request)
Say: "Create the auth module with session, registration, login and logout routes. Start with the anonymous CSRF handshake; registration rotates the cookie and token."
Type: Follow the action above in modules/auth/module.py; explain each new function or binding as you add it.
Viewers should see: The saved file matches this step. After all steps, run the chapter checks below and show the expected result.
Step 5: EDIT - replace the marked blocks - app.py
The application factory mounts only the features already introduced. Follow the action label: a marked-block edit keeps the surrounding factory intact.
Edit 1: find this exact block
Open app.py. Locate this block once. The unchanged surrounding lines identify its position; do not use line numbers from another revision.
from pathlib import Path
from urllib.parse import urlsplit
from flaxon.middleware import BodyLimitMiddleware, TrustedHostsMiddleware
from flaxon import Flaxon, Request
from flaxon.http import JSONResponse
from modules.welcome.module import welcome
from settings import (
ROOT,
Replace that block with
from pathlib import Path
from urllib.parse import urlsplit
from flaxon.middleware import BodyLimitMiddleware, TrustedHostsMiddleware
from argon2 import PasswordHasher
from flaxon import Flaxon, Request
from flaxon.http import JSONResponse
from modules.auth.module import auth
from modules.welcome.module import welcome
from settings import (
ROOT,
Leave the rest of the file unchanged.
Edit 2: find this exact block
Open app.py. Locate this block once. The unchanged surrounding lines identify its position; do not use line numbers from another revision.
TrustedHostsMiddleware, allowed_hosts=[urlsplit(PUBLIC_ORIGIN).hostname]
)
app.public_origin = PUBLIC_ORIGIN
for module, prefix in [
(welcome, "/api/welcome"),
]:
app.mount_module(module, prefix=prefix)
Replace that block with
TrustedHostsMiddleware, allowed_hosts=[urlsplit(PUBLIC_ORIGIN).hostname]
)
app.public_origin = PUBLIC_ORIGIN
app.hasher = PasswordHasher()
app.dummy_password_hash = app.hasher.hash("dummy-password-for-timing-only")
for module, prefix in [
(auth, "/api/auth"),
(welcome, "/api/welcome"),
]:
app.mount_module(module, prefix=prefix)
Leave the rest of the file unchanged.
Say: "The application factory mounts only the features already introduced. Follow the action label: a marked-block edit keeps the surrounding factory intact."
Type: Follow the action above in app.py; explain each new function or binding as you add it.
Viewers should see: The saved file matches this step. After all steps, run the chapter checks below and show the expected result.
Run and check
python management.py runserver
# In a second terminal:
curl -c cookies.txt http://127.0.0.1:8000/api/auth/session
Make the first authenticated request
Keep the server running. In a second terminal use an HTTP client to GET /api/auth/session and save its Set-Cookie value and data.csrf. Then POST /api/auth/register with that cookie, Content-Type: application/json and X-CSRF-Token set to data.csrf. The complete Python demo in chapter 5 automates this exchange. For curl on macOS/Linux:
curl -c cookies.txt http://127.0.0.1:8000/api/auth/session
# Replace TOKEN with data.csrf from the response above:
curl -b cookies.txt -c cookies.txt -H 'Content-Type: application/json' -H 'X-CSRF-Token: TOKEN' -d '{"name":"Learner","email":"learner@example.test","password":"LearningPython123!"}' http://127.0.0.1:8000/api/auth/register
Save the NEW token returned by registration. Do not commit cookies.txt.
Expected result
GET /api/auth/session returns data.user = null and a CSRF token. POST /api/auth/register with the cookie and token creates a customer. GET /api/projects is introduced next; the auth module itself now supports login and logout.
Common errors
Use PUBLIC_ORIGIN exactly, including scheme, hostname and port. A 403 can mean a missing or rotated CSRF token, wrong Origin or non-JSON body. Passwords must have 12-128 characters. Never put the HttpOnly cookie in localStorage.
Short exercise
Use curl or a small HTTP client to register, then log out and show that the previous cookie cannot authenticate.
What to say and show
Say: "Register, sign in, sign out and reject requests without valid sessions. The server remains responsible for persistence and authorization; the browser presents the result."
Show the expected result above, then point to the file responsible for it. Run the chapter check before the next lesson.
Save your checkpoint
git add .
git commit -m "Complete chapter 04: Customer authentication and sessions"