· Patryk Komow · 13 min read

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.

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:

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:

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.

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:

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

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.