Project: a FastAPI service with accounts

FastAPI does the same with get_db: it runs the dependency up to yield, passes the session to the route, and resumes the function once the route is done, so the finally block closes the session. If the route raises an error, FastAPI throws that error into the dependency at the yield, and finally still runs. That's why the try/finally is there: without it, every failed request would leave a session open, and on a busy server those open sessions pile up until the connection pool runs dry and every new request waits.

Passwords and tokens

Signing up stores a bcrypt hash of the password, never the password itself, using a cost (the number of rounds) that makes each guess slow. Logging in checks the password with bcrypt.checkpw and hands back a token: the signed HS256 kind from the auth lesson, carrying the member's id (sub), role and an expiry about 30 minutes away.

The signing secret comes from the environment, not the source code. Code is shared, copied and committed; a secret written in it is a secret every reader of the repository has. The cell below falls back to a practice value only so it runs here.

import base64, hashlib, hmac, json, os, time
import bcrypt

SECRET = os.environ.get("TOKEN_SECRET", "practice-only-secret").encode()
ROUNDS = 4          # fast for the browser; real services use 12 or more

def hash_password(password):
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt(rounds=ROUNDS))

def check_password(password, hashed):
    return bcrypt.checkpw(password.encode(), hashed)

def b64(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

def make_token(member_id, role, minutes=30, now=None):
    now = time.time() if now is None else now
    head = b64(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
    body = b64(json.dumps({"sub": str(member_id), "role": role, "exp": int(now + minutes * 60)}).encode())
    sig = b64(hmac.new(SECRET, f"{head}.{body}".encode(), hashlib.sha256).digest())
    return f"{head}.{body}.{sig}"

def read_token(token, now=None):
    """The token's claims, or None if it's malformed, forged or expired."""
    now = time.time() if now is None else now
    try:
        head, body, sig = token.split(".")
        good = b64(hmac.new(SECRET, f"{head}.{body}".encode(), hashlib.sha256).digest())
        if not hmac.compare_digest(sig.encode(), good.encode()):
            return None
        claims = json.loads(base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)))
    except (ValueError, json.JSONDecodeError):
        return None
    return claims if claims.get("exp", 0) > now else None

h = hash_password("pine-gap-ridge")
print(check_password("pine-gap-ridge", h), check_password("pine-gap-rigde", h))
t = make_token(7, "member")
print(read_token(t)["sub"], read_token(t[:-2] + "xx"), read_token(make_token(7, "member", minutes=-1)))

The last line shows the three outcomes that matter: a good token gives its claims; a tampered signature and an expired token both give None. A route never has to know which went wrong: either way the answer is 401.

Compared with the auth lesson's checklist, two checks are left out on purpose. This service never reads alg from the token: it always verifies with HS256, so a forged header can't switch the algorithm. And one service issues and reads its own tokens, so there's no audience to check; add aud when several services share a secret, and use a maintained library such as PyJWT in production. One more habit: a failed login gets the same 401 whether the email is unknown or the password is wrong, so the login form can't be used to test who's a member. That isn't airtight (signup's 409 says an email is taken, and skipping bcrypt for an unknown email answers faster), which is why real services also limit how often these routes can be called.

Who is asking, and what may they see

get_current_user is the dependency every protected route lists. It turns a token into a member, or stops the request with 401 and the header WWW-Authenticate: Bearer, which tells clients which kind of credential to send. OAuth2PasswordBearer(tokenUrl="token") reads the token from the Authorization header, and records in the API's generated docs that these routes need one.

Authorization is the next step, and the most common serious bug in real APIs is skipping it: a route that loads /trips/42 without checking that trip 42 belongs to the caller. Anyone logged in can then read anyone's data by counting through ids, a hole called an insecure direct object reference (IDOR). The fix is to filter by owner in the query itself, and to answer 404 rather than 403 for someone else's trip, so the API doesn't even confirm that the trip exists.

from fastapi.security import OAuth2PasswordBearer

oauth2 = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2), db: Session = Depends(get_db)) -> Member:
    claims = read_token(token)
    member = db.get(Member, int(claims["sub"])) if claims else None
    if member is None:
        raise HTTPException(401, "Not logged in", headers={"WWW-Authenticate": "Bearer"})
    return member

@app.get("/trips/{trip_id}")
def get_trip(trip_id: int, me: Member = Depends(get_current_user), db: Session = Depends(get_db)):
    trip = db.scalar(select(Trip).where(Trip.id == trip_id, Trip.owner_id == me.id))
    if trip is None:
        raise HTTPException(404, "Trip not found")
    return trip

Notice that owner_id == me.id sits inside the query: it isn't an afterthought checked once the trip is loaded, so a later edit can't easily drop it.

from fastapi import Depends, FastAPI, Header, HTTPException

TRIPS = {1: {"name": "Fern Hollow loop", "owner": "ana"}, 2: {"name": "Red Mesa climb", "owner": "bo"}, 3: {"name": "Pine Gap walk", "owner": "ana"}}
TOKENS = {"t-ana": {"user": "ana", "role": "member"}, "t-bo": {"user": "bo", "role": "member"}, "t-cy": {"user": "cy", "role": "admin"}}

app = FastAPI()


def current_user(authorization: str = Header(default="")):
    user = TOKENS.get(authorization.removeprefix("Bearer "))
    if user is None:
        raise HTTPException(401, "Not logged in", headers={"WWW-Authenticate": "Bearer"})
    return user


@app.get("/trips")
def list_trips(me: dict = Depends(current_user)):
    return [{"id": i, **t} for i, t in TRIPS.items()]


@app.delete("/trips/{trip_id}", status_code=204)
def delete_trip(trip_id: int, me: dict = Depends(current_user)):
    if trip_id not in TRIPS:
        raise HTTPException(404, "Trip not found")
    del TRIPS[trip_id]

Tests that start clean

A test that depends on what an earlier test left in the database passes or fails depending on the order you run them in, the most frustrating kind of flaky test. FastAPI's app.dependency_overrides fixes this: map the real get_db to a replacement, and every route that asks for get_db gets the replacement instead. Give each test a brand-new in-memory database and nothing leaks between them:

def fresh_app():
    sessions = make_sessions()                     # a new, empty in-memory database
    def test_db():
        db = sessions()
        try:
            yield db
        finally:
            db.close()
    app.dependency_overrides[get_db] = test_db
    return app

The routes don't change at all: they still say Depends(get_db). That's the payoff of getting the session from a dependency instead of a global: the same seam lets tests swap the database, and lets production swap SQLite for PostgreSQL. Remember to clear the overrides afterward (app.dependency_overrides.clear()) if other tests use the same app.