Pybites Platform

Back

Fastapi authentication with jwt (json web tokens)

Pybites

Level: Advanced (score: 10)

Up to this point anyone can modify food entries in the food log API, for any user!

That is not desired and in this Bite we will have you lock down the API.

You'll work through the OAuth2 with Password (and hashing), Bearer with JWT tokens guide adding all the pieces to effectively make the API secure using using JWT tokens and secure password hashing. You'll use FastAPI's elegant Depends() to lock down the endpoints that were copied over from a previous Bite.

Good luck!

This Bite concludes the FastAPI learning path. Stay tuned for an SQLModel learning path where we build a similar API but using an ORM / database for persistence.

If you have any feedback on this learning path use the feedback form upon completing any of the Bites or hit us up directly on Circle.

FastAPI

Code
Tests

from datetime import datetime, timedelta
from typing import Any, Dict, List

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# normally would load from env
SECRET_KEY = (
    "963456d8424a9e506d82d1947774c56a2fa3cf1099315cd93e07f44dc5eea21a"  # noqa S105
)
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str

class Food(BaseModel):
    id: int
    name: str
    serving_size: str
    kcal_per_serving: int
    protein_grams: float
    fibre_grams: float = 0

class User(BaseModel):
    id: int
    username: str
    password: str

def __init__(self, **data: Any):
        data["password"] = get_password_hash(data["password"])
        super().__init__(**data)

class FoodEntry(BaseModel):
    id: int
    user: User
    food: Food
    date_added: datetime = datetime.now()
    number_servings: float

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()
food_log: Dict[int, FoodEntry] = {}

# We created an extra in memory user DB
users_db: Dict[str, User] = {}

def verify_password(plain_password, hashed_password):
    """Provided, all good"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    """Provided, all good"""
    return pwd_context.hash(password)

def get_user(username: str):
    """Provided, all good"""
    if username in users_db:
        user = users_db[username]
        return user

def authenticate_user(username: str, password: str):
    """
    TODO: complete this function, use:
    https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/
    """

def create_access_token(data: dict, expires_delta: timedelta):
    """TODO: complete this function"""

def get_current_user(token: str = Depends(oauth2_scheme)):
    """TODO: complete this function"""

@app.post("/create_user", status_code=201)
async def create_user(user: User):
    """Ignore / don't touch this endpoint, the tests will use it"""
    users_db[user.username] = user
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    """TODO: complete this endpoint"""

# TODO: use fastapi.Depends (already imported) to add authentication tot the following endpoints ...

@app.post("/", status_code=201)
async def create_food_entry(entry: FoodEntry):
    food_log[entry.id] = entry
    return entry

@app.get("/", response_model=List[FoodEntry])
async def get_foods_for_user():
    """
    This endpoint does not take a user_id anymore, make it so that the
    food entries are filtered on logged in user.
    """
    return [
        food_entry
        for food_entry in food_log.values()
        # filter by logged in user
    ]

@app.put("/{entry_id}", response_model=FoodEntry)
async def update_food_entry(entry_id: int, new_entry: FoodEntry):
    if entry_id not in food_log:
        raise HTTPException(status_code=404, detail="Food entry not found")

food_log[entry_id] = new_entry

return new_entry

@app.delete("/{entry_id}", response_model=Dict[str, bool])
async def delete_food_entry(entry_id: int):
    if entry_id not in food_log:
        raise HTTPException(status_code=404, detail="Food entry not found")

del food_log[entry_id]

return {"ok": True}

import pytest
from fastapi.testclient import TestClient

from app import FoodEntry, app

LAME_PASSWORD = "1234"  # noqa S105

@pytest.fixture(scope="session")
def client():
    client = TestClient(app)
    return client

@pytest.fixture(scope="session")
def user1():
    user = dict(id=1, username="tim", password=LAME_PASSWORD)
    return user

@pytest.fixture(scope="session")
def user2():
    user = dict(id=2, username="sara", password=LAME_PASSWORD)
    return user

@pytest.fixture(autouse=True, scope="session")
def insert_user(client, user1, user2):
    """Insert users in user DB"""
    for usr in (user1, user2):
        client.post("/create_user", json=usr)

@pytest.fixture(scope="session")
def food1():
    food = dict(
        id=1,
        name="egg",
        serving_size="piece",
        kcal_per_serving=78,
        protein_grams=6.2,
        fibre_grams=0,
    )
    return food

@pytest.fixture(scope="session")
def food2():
    food = dict(
        id=2,
        name="oatmeal",
        serving_size="100 grams",
        kcal_per_serving=336,
        protein_grams=13.2,
        fibre_grams=10.1,
    )
    return food

def test_get_food_entry_without_auth(client, user1, food1):
    resp = client.get("/")
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Not authenticated"}

def test_create_food_entry_without_auth(client, user1, food1):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    resp = client.post("/", json=payload)
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Not authenticated"}

def _get_token(client, username):
    payload = {"username": username, "password": LAME_PASSWORD}
    resp = client.post("/token", data=payload)
    return resp

def test_get_token_success(client, user1):
    resp = _get_token(client, user1["username"])
    assert resp.status_code == 200
    data = resp.json()
    assert "access_token" in data
    assert len(data["access_token"]) == 121
    assert data["token_type"] == "bearer"

def test_get_token_wrong_username(client):
    payload = {"username": "fake", "password": LAME_PASSWORD}
    resp = client.post("/token", data=payload)
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Incorrect username or password"}

def test_get_token_wrong_password(client):
    payload = {"username": "tim", "password": "bad"}
    resp = client.post("/token", data=payload)
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Incorrect username or password"}

def test_create_with_auth(client, user1, food1):
    resp = _get_token(client, user1["username"])
    data = resp.json()
    token = data["access_token"]
    headers = {"Authorization": f"Bearer {token}"}
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    resp = client.post("/", json=payload, headers=headers)
    assert resp.status_code == 201
    data = resp.json()
    assert data["id"] == 1
    assert data["user"]["id"] == 1
    assert data["food"]["name"] == "egg"
    assert data["number_servings"] == 1.5

def test_create_with_wrong_token(client, user1, food1):
    headers = {"Authorization": "Bearer fake_token"}
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    resp = client.post("/", json=payload, headers=headers)
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Could not validate credentials"}

def test_can_only_create_own_foods(client, user1, user2, food1):
    resp = _get_token(client, user1["username"])
    data = resp.json()
    token = data["access_token"]
    headers = {"Authorization": f"Bearer {token}"}
    payload = dict(id=1, user=user2, food=food1, number_servings=1.5)
    resp = client.post("/", json=payload, headers=headers)
    assert resp.status_code == 400
    assert resp.json() == {"detail": "Can only add food for current user"}

def _create_food_as_user(client, payload, username):
    resp = _get_token(client, username)
    data = resp.json()
    token = data["access_token"]
    headers = {"Authorization": f"Bearer {token}"}
    resp = client.post("/", json=payload, headers=headers)
    assert resp.status_code == 201
    return headers

def test_get_only_returns_own_foods(client, user1, user2, food1, food2):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=2, user=user2, food=food2, number_servings=2)
    headers = _create_food_as_user(client, payload, user2["username"])
    resp = client.get("/", headers=headers)
    # only seeing user2 foods
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    item = data[0]
    assert item["id"] == 2
    assert item["user"]["id"] == 2
    assert item["food"]["name"] == "oatmeal"
    assert item["number_servings"] == 2.0

def test_update_needs_login(client, user1, food1):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=1, user=user1, food=food1, number_servings=2)
    resp = client.put("/1", json=payload)
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Not authenticated"}

def test_update_wrong_food(client, user1, food1):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    headers = _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=1, user=user1, food=food1, number_servings=2)
    resp = client.put("/3", json=payload, headers=headers)
    assert resp.status_code == 404

def test_update_food_not_owner_by_user(client, user1, user2, food1, food2):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=2, user=user2, food=food2, number_servings=2)
    headers = _create_food_as_user(client, payload, user2["username"])
    # cannot update user1 food as user2
    resp = client.put("/1", json=payload, headers=headers)
    assert resp.status_code == 400
    assert resp.json() == {"detail": "Food entry not owned by you"}

def test_update_food_owned_by_self(client, user1, user2, food1, food2):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    headers = _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=1, user=user1, food=food1, number_servings=2)
    resp = client.put("/1", json=payload, headers=headers)
    assert resp.status_code == 200
    data = resp.json()
    assert data["number_servings"] == 2.0

def test_delete_needs_login(client, user1, food1):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    _create_food_as_user(client, payload, user1["username"])
    resp = client.delete("/1")
    assert resp.status_code == 401
    assert resp.json() == {"detail": "Not authenticated"}

def test_delete_wrong_food(client, user1, food1):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    headers = _create_food_as_user(client, payload, user1["username"])
    resp = client.delete("/3", headers=headers)
    assert resp.status_code == 404

def test_delete_food_not_owner_by_user(client, user1, user2, food1, food2):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    _create_food_as_user(client, payload, user1["username"])
    payload = dict(id=2, user=user2, food=food2, number_servings=2)
    headers = _create_food_as_user(client, payload, user2["username"])
    # cannot delete user1 food as user2
    resp = client.delete("/1", headers=headers)
    assert resp.status_code == 400
    assert resp.json() == {"detail": "Food entry not owned by you"}

def test_delete_food_owned_by_self(client, user1, user2, food1, food2):
    payload = dict(id=1, user=user1, food=food1, number_servings=1.5)
    headers = _create_food_as_user(client, payload, user1["username"])
    resp = client.delete("/1", headers=headers)
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}