FastAPI authentication: sessions, JWT, email verification and password reset
Sign-up, login, logout, a confirmation email, a password reset link and Bearer tokens for an API, built on one user table and three small libraries. No auth service, no extra infrastructure, and every session can be revoked.
Contents
What we're building
Most FastAPI auth tutorials stop at "here's a JWT". A product needs more: HTML pages that remember who's signed in, a way to confirm the email address, a way back in when the password is forgotten, and a way to sign someone out of every device when their password changes. Here's the whole set:
- Browser pages use a signed session cookie. It's what browsers are good at, and it needs no JavaScript.
- The JSON API uses short-lived JWT Bearer tokens.
- Email links (verification, reset) are signed tokens with an expiry, so there's no tokens table to clean up.
- One column, the auth stamp, lets you revoke all of the above at once.
Libraries: argon2-cffi for password hashes, itsdangerous for signed links, PyJWT for API tokens, email-validator for addresses, plus FastAPI, Starlette's SessionMiddleware and async SQLAlchemy. The code is trimmed from a working app.
The user table and the auth stamp
# app/accounts/models.py
def new_auth_stamp() -> str:
return secrets.token_hex(16)
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
# Always stored lowercase, see normalize_email().
email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
password_hash: Mapped[str] = mapped_column(String(255))
is_active: Mapped[bool] = mapped_column(default=True)
email_verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
# Rotated when the password changes. Sessions, API tokens and reset links carry
# the stamp they were issued with, so rotating it signs the user out everywhere.
auth_stamp: Mapped[str] = mapped_column(String(32), default=new_auth_stamp)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
@property
def is_verified(self) -> bool:
return self.email_verified_at is not None
The auth_stamp is the trick that makes stateless sessions revocable. A signed cookie or a JWT can't be "deleted" on the server, because the server doesn't store them. But if each one carries the stamp it was issued with, and every request compares that with the stamp in the database, then changing the stamp invalidates every cookie and token issued before. One column, no sessions table.
Store email_verified_at as a timestamp rather than a boolean. It costs nothing and answers "since when?" when you need it.
Passwords: Argon2, off the event loop
# app/accounts/security.py
from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError
_hasher = PasswordHasher()
# Verified against when the email doesn't exist, so a failed login takes the same
# time whether or not the account exists.
_DUMMY_HASH = _hasher.hash("dummy password for timing")
def hash_password(password: str) -> str:
return _hasher.hash(password)
def verify_password(password_hash: str | None, password: str) -> bool:
"""Pass None when the user doesn't exist: the check still runs."""
try:
_hasher.verify(password_hash or _DUMMY_HASH, password)
except (VerificationError, InvalidHashError):
return False
return password_hash is not None
def password_needs_rehash(password_hash: str) -> bool:
"""True after you raise the Argon2 cost parameters; rehash on the next login."""
return _hasher.check_needs_rehash(password_hash)
Three details worth copying:
- Argon2id is the default in
argon2-cffiand the current recommendation for new code. Its defaults are sensible; don't tune them unless you've measured. - The dummy hash closes a timing leak. Without it, "unknown email" returns in a millisecond and "wrong password" in tens of milliseconds, which tells an attacker which emails have accounts.
check_needs_rehashlets you raise the cost later: old hashes are upgraded the next time each user logs in, when you briefly have the plain password.
Argon2 is deliberately slow and CPU-bound. Called directly in an async route, it blocks the event loop and every other request waits. Run it in a thread:
# app/accounts/service.py
async def authenticate(session: AsyncSession, email: str, password: str) -> User | None:
"""Returns the user if the email and password match an active account."""
user = await get_user_by_email(session, email)
ok = await asyncio.to_thread(
verify_password, user.password_hash if user else None, password
)
if not ok or user is None or not user.is_active:
return None
if password_needs_rehash(user.password_hash):
user.password_hash = await asyncio.to_thread(hash_password, password)
await session.commit()
return user
Sign-up
def normalize_email(email: str) -> str:
return email.strip().lower()
async def create_user(session: AsyncSession, email: str, password: str) -> User:
password_hash = await asyncio.to_thread(hash_password, password)
user = User(email=normalize_email(email), password_hash=password_hash)
session.add(user)
try:
await session.commit()
except IntegrityError as exc:
await session.rollback()
raise EmailTakenError from exc
return user
Lowercase emails before storing and before every lookup, or Anna@Example.com and anna@example.com become two accounts. And let the unique constraint decide whether an email is taken: a "SELECT, then INSERT" check has a race between two sign-ups with the same address, and catching IntegrityError doesn't.
The route validates the address with email-validator (without a DNS lookup, which is slow and fails on new domains), checks the password length, creates the user, sends the verification email, and logs them in straight away:
@router.post("/register")
async def register(
request: Request,
session: SessionDep,
email: Annotated[str, Form()],
password: Annotated[str, Form()],
):
try:
email = validate_email(email, check_deliverability=False).normalized
except EmailNotValidError:
return fail("auth.error.invalid_email")
if problem := service.password_problem(password): # 8 to 128 characters
return fail(problem)
try:
user = await service.create_user(session, email, password)
except service.EmailTakenError:
return fail("auth.error.email_taken")
await service.send_verification_email(user)
login_session(request, user)
return RedirectResponse("/account", status_code=303)
Logging in before the email is confirmed is a product decision. Making people click a link before they can see anything loses sign-ups; instead, let them in and require a verified email only where it matters (paying, inviting others).
Cap the password length too. Argon2 handles long inputs, but a 1 MB "password" still costs you CPU time.
Sessions in a signed cookie
Starlette's SessionMiddleware stores the session as a signed cookie. Signed, not encrypted: the user can read it, but not change it. So it holds the user id and the stamp, never anything secret.
# app/main.py
app.add_middleware(
SessionMiddleware,
secret_key=settings.secret_key,
session_cookie="session",
max_age=settings.session_max_age_days * 24 * 3600,
same_site="lax",
https_only=settings.is_production,
)
The cookie is HttpOnly by default (scripts can't read it); same_site="lax" keeps it off cross-site POSTs; https_only marks it Secure in production. Logging in and out:
# app/accounts/deps.py
SESSION_USER_KEY = "uid"
SESSION_STAMP_KEY = "stamp"
def login_session(request: Request, user: User) -> None:
# Start from a clean session, so nothing from before the login carries over.
request.session.clear()
request.session[SESSION_USER_KEY] = user.id
request.session[SESSION_STAMP_KEY] = user.auth_stamp
def logout_session(request: Request) -> None:
request.session.clear()
Clearing the session on login matters: whatever an anonymous visitor had in their session (or whatever someone planted there) shouldn't survive into the signed-in one.
A cookie session means the browser sends credentials automatically, so forms need CSRF protection. Put a token in the session, render it in every form (and in an hx-headers attribute if you use HTMX), and check it in a dependency on every router that accepts forms.
Protecting routes with dependencies
One function loads the user for every request and checks the stamp:
async def load_user(request: Request, session: SessionDep) -> User | None:
request.state.user = None
user_id = request.session.get(SESSION_USER_KEY)
if user_id is None:
return None
user = await session.get(User, user_id)
stamp_ok = user is not None and user.auth_stamp == request.session.get(SESSION_STAMP_KEY)
if user is None or not user.is_active or not stamp_ok:
# Deleted, deactivated, or the password changed since this session started.
logout_session(request)
return None
request.state.user = user
return user
async def require_user(user: Annotated[User | None, Depends(load_user)]) -> User:
if user is None:
raise LoginRequiredError
return user
async def require_verified_user(user: Annotated[User, Depends(require_user)]) -> User:
if not user.is_verified:
raise VerificationRequiredError
return user
CurrentUser = Annotated[User, Depends(require_user)]
VerifiedUser = Annotated[User, Depends(require_verified_user)]
Register load_user as an app-wide dependency (FastAPI(dependencies=[Depends(load_user)])) so templates can always show who's signed in. Routes then say what they need in their signature:
@router.get("/projects")
async def projects(request: Request, user: CurrentUser): ...
@router.post("/billing/checkout/{plan_key}")
async def checkout(plan_key: str, session: SessionDep, user: VerifiedUser): ...
The dependencies raise exceptions instead of returning redirects, because a dependency can't return a response. Two exception handlers turn them into redirects, and the login one remembers where the user was going:
@app.exception_handler(LoginRequiredError)
async def login_required(request: Request, exc: LoginRequiredError):
next_path = quote(request.url.path, safe="/")
return RedirectResponse(f"/login?next={next_path}", status_code=303)
@app.exception_handler(VerificationRequiredError)
async def verification_required(request: Request, exc: VerificationRequiredError):
return RedirectResponse("/account", status_code=303)
FastAPI caches a dependency within one request, so load_user runs once even though both the app and require_user depend on it.
Login, rate limits and the next parameter
@router.post("/login")
async def login(
request: Request,
session: SessionDep,
email: Annotated[str, Form()],
password: Annotated[str, Form()],
next_url: Annotated[str, Form(alias="next")] = "",
):
key = f"{client_ip(request)}:{service.normalize_email(email)}"
if login_limiter.is_blocked(key):
return fail("auth.error.too_many_attempts")
user = await service.authenticate(session, email, password)
if user is None:
login_limiter.hit(key)
return fail("auth.error.bad_credentials")
login_limiter.reset(key)
login_session(request, user)
return RedirectResponse(safe_redirect_path(next_url, "/account"), status_code=303)
The error says "wrong email or password", never which one. The rate limit counts failures per IP and email, so one attacker can't lock a real user out from everywhere, and a success resets the count. The limiter itself is a dict of timestamps, a sliding window per key:
class RateLimiter:
def __init__(self, max_attempts: int, window_seconds: int = 15 * 60):
self.max_attempts = max_attempts
self.window_seconds = window_seconds
self._attempts: dict[str, deque[float]] = defaultdict(deque)
def _prune(self, key: str, now: float) -> deque[float]:
attempts = self._attempts[key]
while attempts and attempts[0] <= now - self.window_seconds:
attempts.popleft()
return attempts
def is_blocked(self, key: str) -> bool:
return len(self._prune(key, time.monotonic())) >= self.max_attempts
def hit(self, key: str) -> None:
now = time.monotonic()
self._prune(key, now).append(now)
It lives in memory, so each worker process counts separately. For password guessing that's still a hard limit; if you run many servers and need one shared count, move it to Redis. Behind a reverse proxy, make sure the client IP is the visitor's and not the proxy's (uvicorn's --proxy-headers with FORWARDED_ALLOW_IPS set to your proxy).
The next parameter is a classic open redirect: /login?next=https://evil.example logs the user in and sends them to a phishing page. Only accept local paths:
def safe_redirect_path(value: str | None, default: str = "/") -> str:
"""Only allow local paths, so ?next= can't send users to another site.
Browsers read "/\\" like "//", so both are rejected."""
if not value or not value.startswith("/") or value[1:2] in ("/", "\\"):
return default
return value
//evil.example is a valid "path" that browsers treat as another host, and some browsers read /\evil.example the same way. Hence the second character check.
Email links without a tokens table
A verification or reset link needs to prove three things: we issued it, it hasn't expired, and it's for this purpose. itsdangerous gives you all three without storing anything:
# app/accounts/security.py
from itsdangerous import BadSignature, URLSafeTimedSerializer
# Each purpose gets its own salt, so a verification link can't be used as a reset link.
VERIFY_EMAIL = "verify-email"
RESET_PASSWORD = "reset-password"
def _serializer(purpose: str) -> URLSafeTimedSerializer:
return URLSafeTimedSerializer(get_settings().secret_key, salt=purpose)
def make_link_token(purpose: str, data: dict) -> str:
return _serializer(purpose).dumps(data)
def read_link_token(purpose: str, token: str, max_age_seconds: int) -> dict | None:
"""Returns the data, or None if the token is forged, expired or for another purpose."""
try:
return _serializer(purpose).loads(token, max_age=max_age_seconds)
except BadSignature: # SignatureExpired is a subclass
return None
The verification link carries the user id and the email address it was sent to:
def verification_link(user: User) -> str:
token = make_link_token(VERIFY_EMAIL, {"uid": user.id, "email": user.email})
return f"{get_settings().base_url}/verify-email?token={token}"
async def user_from_verification_token(session: AsyncSession, token: str) -> User | None:
data = read_link_token(VERIFY_EMAIL, token, max_age=48 * 3600)
if data is None:
return None
user = await session.get(User, data["uid"])
# A link sent to an old address must not verify a changed one.
if user is None or user.email != data["email"]:
return None
return user
Without the email in the token, someone could sign up with their own address, change it to someone else's, and then click the old link to "verify" an address they don't own.
Build links from a configured BASE_URL, not from request.url. The Host header comes from the client, and a password reset email that points to the attacker's domain is a well-known way to steal accounts.
Password reset
The reset link carries the auth stamp. That makes it single-use without storing it: resetting the password rotates the stamp, and every older link stops working.
def reset_password_link(user: User) -> str:
# The stamp makes the link single-use: resetting the password rotates it.
token = make_link_token(RESET_PASSWORD, {"uid": user.id, "stamp": user.auth_stamp})
return f"{get_settings().base_url}/reset-password?token={token}"
async def user_from_reset_token(session: AsyncSession, token: str) -> User | None:
data = read_link_token(RESET_PASSWORD, token, max_age=60 * 60)
if data is None:
return None
user = await session.get(User, data["uid"])
if user is None or not user.is_active or user.auth_stamp != data["stamp"]:
return None
return user
async def change_password(session: AsyncSession, user: User, new_password: str) -> None:
"""Sets a new password and signs the user out of every other session and API token."""
user.password_hash = await asyncio.to_thread(hash_password, new_password)
user.auth_stamp = new_auth_stamp()
await session.commit()
The "forgot password" form answers the same way whether or not the email has an account, so it can't be used to check who's a customer:
@router.post("/forgot-password")
async def forgot_password(request: Request, session: SessionDep, email: Annotated[str, Form()]):
key = f"reset:{client_ip(request)}"
if email_limiter.is_blocked(key):
return fail("auth.error.too_many_attempts")
email_limiter.hit(key)
user = await service.get_user_by_email(session, email)
if user and user.is_active:
await service.send_password_reset_email(user)
# Same answer either way, so the form can't be used to find out who has an account.
flash(request, "auth.flash.reset_sent")
return RedirectResponse("/login", status_code=303)
Rate-limit it, or it becomes a free way to send email to anyone from your domain. After a successful reset, also mark the email as verified (opening the link proved they own the inbox) and log the user in, since they've just proved who they are.
The same rotation gives you a "sign out of all devices" button for free: generate a new stamp, commit, and log the current browser back in with login_session().
JWT for the JSON API
Mobile apps and scripts can't use a cookie session comfortably, so the API gets Bearer tokens. FastAPI's OAuth2PasswordBearer reads the header and wires up the "Authorize" button in /docs:
# app/accounts/security.py
def create_access_token(user_id: int, auth_stamp: str) -> str:
settings = get_settings()
now = datetime.now(UTC)
payload = {
"sub": str(user_id),
"stamp": auth_stamp,
"iat": now,
"exp": now + timedelta(minutes=settings.access_token_minutes),
}
return jwt.encode(payload, settings.secret_key, algorithm="HS256")
def decode_access_token(token: str) -> dict | None:
try:
return jwt.decode(token, get_settings().secret_key, algorithms=["HS256"])
except jwt.PyJWTError:
return None
# app/accounts/deps.py
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token", auto_error=False)
async def require_api_user(
session: SessionDep, token: Annotated[str | None, Depends(oauth2_scheme)]
) -> User:
unauthorized = HTTPException(
status_code=401, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}
)
payload = decode_access_token(token) if token else None
if payload is None:
raise unauthorized
user = await session.get(User, int(payload["sub"]))
if user is None or not user.is_active or user.auth_stamp != payload.get("stamp"):
raise unauthorized
return user
ApiUser = Annotated[User, Depends(require_api_user)]
Two things that trip people up:
- Pass
algorithms=[...]tojwt.decode, as a list you control. Accepting whatever algorithm the token's header names is how "alg: none" attacks work. submust be a string. Recent PyJWT versions reject an integersubas invalid, so convert on the way in and back on the way out.
Because the token carries the stamp too, changing the password revokes API tokens along with browser sessions. The token endpoint uses OAuth2PasswordRequestForm (the email goes in the username field, per the OAuth2 spec) and shares the login rate limiter:
curl -X POST localhost:8000/api/auth/token -d "username=you@example.com&password=..."
curl localhost:8000/api/me -H "Authorization: Bearer eyJhbGciOi..."
Checklist
- Argon2id, run in a thread, with a dummy hash for unknown emails and rehash on login.
- Emails lowercased; the unique constraint decides whether one is taken.
- Signed-cookie session holding only the user id and the auth stamp; cleared on login.
- CSRF tokens on every form, because cookies are sent automatically.
CurrentUserandVerifiedUserdependencies, with exception handlers for the redirects.- Login rate limit per IP and email;
nextlimited to local paths. - Email links signed with a salt per purpose and an expiry; the verification link bound to the address, the reset link to the stamp.
- Links built from
BASE_URL, never from the Host header. - The same answer from "forgot password" whether or not the account exists.
- JWT with a fixed algorithm list, a string
sub, and the stamp.
This is the accounts module from getpylaunch, a FastAPI SaaS starter I sell, where it comes with the templates, emails in two languages and the tests already written.