TDD Approach to Create an Authentication System With FastAPI Part 4

Introduction

In the last couple of posts in TDD Auth with FastAPI we’ve been sustainably moved towards a web service that can let users register with the service. Now what?

We have already done the easy part. The next part is to look at the authorization. This involves letting the user log in. And only give access to what they are authorized for. Let us look at the login part first.

Before talking into login, I’d like to tell you that I have always found authentication and authorization challenging. There are so many terminologies involved. A few to name are OAuth, Basic Authentication, JWT, Cookies, SAML, and whatnot. I can’t explain all of them in this single post, but I’ll try to cover some basic building blocks to get you started.

Series Index

We have already seen:

Authentication Premier

What is Authentication?

Authentication in most basic terms is the process of validating an identity to ascertain that they are who they claim to be. By the way, authentication can be achieved using passwords, OTPs, biometrics, authentication apps, access tokens, certificates, and more. We’ll be authenticating using passwords and tokens.

Following is a sequence diagram that roughly describes what happens in a successful login process.

Once you are authenticated, the server provides a token, this token then needs to be sent to the server in each request by the client. you are authorized to do certain activities on the application.

Different kinds of authentication (and scenario)

There are mainly two kinds of application authenticating.

User to application authentication. When I say user to application authentication, I refer to a scenario where a user (using a client, like a web browser) is logging into an application.
Application to application authentication. App to app authentication is when you use one service (say Google) to sign in to another application (say appX). The “Signin with ” button represents an example of app-to-app authentication.

Here you can argue that at the end of the day user is going to be authenticated so this is a user to app authentication. But my point is, that we are being authenticated to an application, using the data from another application.

We are not talking about an app to app authentication in this post. For this post, I’ll only be exploring simple authentication where a user uses their username and password. And when username and password match, an authentication token is sent to the frontend. Their token is required to be identified as logged in and to be able to access restricted resources.

In this section we talk about some terminologies and how are they related to each other.

HTTP and the Authorization Header

You must be knowing about different request and response headers when dealing with HTTP in general. Some of the common ones which are on top of my mind are:

Request header: User-Agent, Referer, Origin, Host, Content-Type, Content-Length, Accept.
Response header: Status, Location, Content-Type, Cache-Control.

Among those request headers, there is one more header called Authorization. If you are dealing with auth, the first thing you must do is learn more about this header. I will cover some of the basic info here and give you some links at the end of this section for further reading.

An Authorization header looks like this:

Authorization: <scheme> <credentials>

<scheme> is something we’ll see in the next section. <credentials> are a piece of string, the value of which will depend on what the scheme is.

https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Authorization

Authentication Schemes

There are over half a dozen of authentication schemes, but these ones are common.

Basic
Bearer
Digest

Basic

Basic authentication scheme deals with username and password. In basic HTTP authentication, a request contains a header field in the form of:

Authorization: Basic username:password

where credentials are base64 encoded so the actual header will look like this:

Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

Some final points:

The security of the Basic scheme is only dependent on the transport layer i.e. HTTPS.
HTTP does not provide a method for a web server to instruct the client to “log out” the user.
You should not be using the Basic scheme. There is a similar approach taken in the Digest scheme which you might consider.

Bearer

Originally taken from OAuth, Bearer deals with tokens. The name “Bearer authentication” can be understood as “give access to the bearer of this token”. In terms of the header, the Bearer is not much different from Basic. It looks like

Authorization: Bearer <credentials>

But instead of passing bare username:password in credentials, they are signed and encrypted. They are more secure than Basic because these tokens can have an expiry time set to them. When the server reads the token passed from a client and finds that it has expired, it marks the request as unauthorized. In that case, the client again has to generate the token.

Final points:

Although Bearer is safer than Basic, it is highly advised to use them over HTTPS.
We are going to discuss Bearer tokens only in this post. We’ll use JSON Web Token. If you are feeling crazy, read the spec of JWT to know about it in-depth.
Bearer scheme in my opinion is the most widely used scheme on the internet.

https://stackoverflow.com/questions/34013299/web-api-authentication-basic-vs-bearer

Digest

Simply speaking, Digest is a complex and secure version of the Basic scheme. I have not used Digest at the moment and thus I’m not going to talk about it in this post. But if you want to read about it, please follow https://en.wikipedia.org/wiki/Digest_access_authentication.

Define requirement for authentication (/users/auth) endpoint

We have already seen in the above diagram that username and password go from the frontend application to the backend service endpoint. The client sending this password could be a web browser like Firefox, a CLI tool like curl, or even an API testing tool like Postman.

Now let us describe the functional requirement of our /users/auth endpoint. Thinks about what our endpoints will look like in a finished product. Here is what I can think of:

The consumer hits /users/auth with the GET method and should get Method Not Allowed.
The API consumer hits /users/auth with a POST method:
- Without body = returns unauthorized
- With body:
  - If user not found OR username exists but the password is wrong, returns unauthorized
  - Should check in the database if the user exists, if so, creates and returns the JWT token

Functional requirements turned into test

Below are the test cases for the spec. This is a slightly modified version of the registration cases.

tests/test_users.py

53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@@ -54,3 +54,32 @@ class TestUserRegistration:
             json={"username": "santosh", "password": "sntsh", "fullname": "Santosh Kumar"}
         )
         assert response.status_code == 201
+
+
+class TestUserLogin:
+    """TestUserLogin tests /users/auth"""
+
+    def test_get_request_returns_405(self):
+        """login endpoint does only expect a post request"""
+        response = client.get("/users/auth")
+        assert response.status_code == 405
+
+    def test_post_request_without_body_returns_422(self):
+        """body should have username, password and fullname"""
+        response = client.post("/users/auth")
+        assert response.status_code == 422
+
+    def test_post_request_with_improper_body_returns_422(self):
+        """both username and password is required"""
+        response = client.post(
+            "/users/auth",
+            json={"username": "santosh"}
+        )
+        assert response.status_code == 422
+
+    def test_post_request_with_proper_body_returns_200_with_jwt_token(self):
+        response = client.post(
+            "/users/auth",
+            json={"username": "santosh", "password": "sntsh"}
+        )
+        assert response.status_code == 200
+        assert len(response.json()) == 2

And when you run these tests, they’re obviously gonna throw 404.

$ pytest --tb=line
============================== test session starts ==============================
platform linux -- Python 3.8.10, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /efs/repos/fastauth
plugins: anyio-3.4.0
collected 9 items                                                                                                          

tests/test_main.py .                                                   [ 11%]
tests/test_users.py ....FFFF                                           [100%]

==================================== FAILURES ===================================
/efs/repos/fastauth/tests/test_users.py:65: assert 404 == 405
/efs/repos/fastauth/tests/test_users.py:70: assert 404 == 422
/efs/repos/fastauth/tests/test_users.py:78: assert 404 == 422
/efs/repos/fastauth/tests/test_users.py:85: assert 404 == 201

Our tests are red. We’ll have to do something to turn them green.

Write code to fix failing test

Add new models for credential data and token

User has to send their username and password to be able to log in to the system and get a token. Here are the changes:

fastauth/schemas.py

19
20
21
22
23
24
25
26
27
28
+
+
+class UserAuthenticate(BaseModel):
+    username: str
+    password: str
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str

UserAuthenticate indicates a structure of JSON that we are supposed to receive in a request from a user.

Token denotes a structure that the server will return.

Define /users/auth route

Let’s have a route as per our spec to let a user log in. This route, after receiving proper credentials, returns a JWT token

fastauth/main.py

11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 from fastapi import FastAPI, Depends, HTTPException
 from sqlalchemy.orm import Session
 
-from fastauth import schemas, crud
+from fastauth import schemas, crud, auth
 from fastauth.database import DBInit
 
+# These constants go in a specific config file
+ACCESS_TOKEN_EXPIRE_MINUTES=15
 
 def get_db():
     session = None
@@ -20,9 +22,26 @@ app = FastAPI()
 async def ping():
     return {'msg': 'pong'}
 
[email protected]("/users/register", status_code=201, response_model=schemas.UserInfo)
[email protected]("/users/register", status_code=201, response_model=schemas.UserInfo, tags=["users"])
 def register_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
     db_user = crud.get_user_by_username(db, username=user.username)
     if db_user:
         raise HTTPException(status_code=409, detail="Username already registered")
     return crud.create_user(db=db, user=user)
+
+
[email protected]("/users/auth", response_model=schemas.Token, tags=["users"])
+def authenticate_user(user: schemas.UserAuthenticate, db: Session = Depends(get_db)):
+    db_user = crud.get_user_by_username(db, username=user.username)
+    if db_user is None:
+        raise HTTPException(status_code=403, detail="Username or password is incorrect")
+    else:
+        is_password_correct = auth.check_username_password(db, user)
+        if is_password_correct is False:
+            raise HTTPException(status_code=403, detail="Username or password is incorrect")
+        else:
+            from datetime import timedelta
+            access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+            access_token = auth.encode_jwt_token(
+                data={"sub": user.username}, expires_delta=access_token_expires)
+            return {"access_token": access_token, "token_type": "Bearer"}

Let us start from line 33 where we define the route /users/auth. response_model says that this endpoint in good terms returns a schemas.Token. tags, which is also on 25 is to group endpoint on Swagger UI.

On line 34, we see that authenticate_user accepts user which comes from a request, and db which is a DB session.

On lines 35-37, we check if the passed user exists or not, if not, HTTP 403 is returned. If that is not the case, the function continues to run.

If the user exists, we check if the password matches with the help of a helper function check_username_password. If it does not, we again throw a 403.

The only case that remained now is to return a token. On lines 43-47, we create a token with an expiry set to ACCESS_TOKEN_EXPIRE_MINUTES minutes. After this much time, the token expires. We are again using a helper function called encode_jwt_token.

Auth helper function

Now let us see what helper function we have for aiding in the authentication.

Before writing this file, please install PyJWT.

pip install PyJWT

fastauth/auth.py

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import datetime

import bcrypt
import jwt
from sqlalchemy.orm import Session

from . import models, schemas, crud


SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
JWT_ALGORITHM = "HS256"

def check_username_password(db: Session, user: schemas.UserAuthenticate):
    db_user_info: models.UserInfo = crud.get_user_by_username(db, username=user.username)
    db_pass = db_user_info.password.encode('utf8')
    request_pass = user.password.encode('utf8')
    return bcrypt.checkpw(request_pass, db_pass)

def encode_jwt_token(*, data: dict, expires_delta: datetime.timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.datetime.utcnow() + expires_delta
    else:
        expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
    to_encode.update({"exp": expire})

    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=JWT_ALGORITHM)
    return encoded_jwt

In check_username_password, we are checking the received password with the password stored in the database. Of course, the database has the hashed version of the password store. This is one of the best practices we must incorporate. Please don’t store passwords in plain text.

In encode_jwt_token, we use the PyJWT module to create a JWT token with 15 minutes of expiry. The code is pretty much self-explanatory.

Fixing create_user

I just noticed that I need to decode the output of bcrypt.hashpw before saving it into the database.

1
2
3
4
5
6
7
  def create_user(db: Session, user: schemas.UserCreate):
-    hashed_password = bcrypt.hashpw(user.password.encode('utf-8'), bcrypt.gensalt())
+    hashed_password = bcrypt.hashpw(user.password.encode('utf8'), bcrypt.gensalt())
+    hashed_password = hashed_password.decode('utf8')
     db_user = models.UserInfo(username=user.username, password=hashed_password, fullname=user.fullname)
     db.add(db_user)
     db.commit()

Confirm if tests are passing

Automated tests

Let us run our test suite

$ pytest
==================== test session starts =====================
platform linux -- Python 3.8.10, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /efs/repos/fastauth
plugins: anyio-3.4.0
collected 9 items                                            

tests/test_main.py .                                   [ 11%]
tests/test_users.py ........                           [100%]

===================== 9 passed in 2.17s ======================

Automated tests are in a happy state.

Test manually using SwaggerUI

Let’s test our API manually also.

With this, let’s end this post here. In the next post, we shall implement a protected route and shall see how we can use our so-called JWT token to access those routes. Until then, have a nice day.

Conclusion

In this post, we saw some of the very basic concepts to get you started with a login system. We saw how HTTP and Authorization header is involved in this. We also saw a different types of logins. We talked about Authentication Schemes.

At last, we came up with a login endpoint spec and turned it into tests, next we implemented the test. Kudos to you!

Please join my subscriber list to get updates on my blog. Please enter your email below.