JWT Auth-101
How to handle authentication between different services?
Nowadays I am doing research about microservices. Actually, I understood some of the keypoints. However, there was a foggy point like “how I can handle authentication for my post service by using auth service”. In this text, we are going to see how we can handle authentication between different systems with applied examples in Python, Flask.
What is Authentication and Authorization?
Authentication is the process of identifying whether a user exists in our system or not.
Authorization is the process of determining whether a user has permission to perform a specific action or access a particular resource.
In other words, after a user has been authenticated and their identity has been confirmed, authorization is the process of checking if that user is allowed to perform a certain action or access a specific resource. For example, a user might be authenticated and logged into a system, but they might not be authorized to access certain sensitive information or perform certain actions, such as modifying critical system settings.
What is JWT?
JWT is a compact and self-contained way of transmitting information between two parties. It is typically used for authentication and authorization, and allows the parties to securely exchange information such as user identities and other claims. JWT typically consists of a JSON object that is digitally signed and optionally encrypted, and is typically sent as a bearer token in the HTTP Authorization header.
“If you are hearing about JWT for the first time, you can visit jwt.io to learn more about it.”
JWT Authentication in Theory
- The user send a request to login endpoint of auth service.
- I use “HTTP Basic” to send a login request, you can use request body or other methods.
2. Service parse the Authorization header of the request and checks if user is exists in the system.
- If there is no user, return error message and 401 status code.
- If the user is exits in the system, create JWT and return the token.
3. Get action by the response.
- If there is token, put the token in HTTP Authorization header (Bearer <TOKEN>).
4. To request other services by using the token.
JWT Authentication in Action
In this section, we’ll explore the theory in action by using Flask which is a web framework of Python ecosystem.
Requirements
We are going to use the following packages in the project.
flask: Web server (pip install flask)pyjwt: Generating tokens (pip install pyjwt)httpx: Making requests to the endpoints (pip install httpx)
Auth Service
Firstly, I am going to define the constants.
"""
# We use the snippets below to make requests
# login post request
- httpx -m POST --auth user@user.com Passw0rd http://127.0.0.1:5000/login
# validate post request
- httpx -m POST -h "Authorization" "Bearer <TOKEN>" http://127.0.0.1:5000/validate
"""
from flask import Flask, request
import jwt
import datetime
JWT_SECRET = (
"this-is-so-secret-key-to-encode-jwt-tokens" # you should get this by using ENV
)
USERS_DB = [
{"username": "user@user.com", "password": "Passw0rd", "is_admin": True}
] # you should use a real db :)
app = Flask(__name__)app will reference the server.
JWT_SECRET will be used in the JWT creating process.
USERS_DB will be used as a dummy database. I chose this method to keep it simple.
@app.route("/login", methods=["POST"])
def login():
print(request.headers) # We need to get "Authorization" header of the request
print(request.authorization) # This is automatically decode the header in Flask
# handle username and password from request header (Flask does for us, it decodes)
auth = request.authorization
if not auth:
return "missing credentials", 401
# filter users
users_list = [u for u in USERS_DB if u["username"] == auth.username]
# is there any user with the credentials
if len(users_list) != 1 or users_list[0]["password"] != auth.password:
return ("Missing credentials", 401)
else:
return createJWT(auth.username, JWT_SECRET, users_list[0]["is_admin"]), 200As I mentioned earlier, we need to handle Authorization header of the request. To do that, we can use request.headers from the Flask. I am going to send a POST request by using httpx for simulating the code above.
httpx -m POST --auth user@user.com Passw0rd http://127.0.0.1:5000/loginAnd we’ll get an output like below.
Host: 127.0.0.1:5000
Content-Length: 0
Accept: */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
User-Agent: python-httpx/0.23.1
Authorization: Basic dXNlckB1c2VyLmNvbTpQYXNzdzByZA==As you see the Authorization header, we get a encoded value.
You can check RFC 7617 - The Basic HTTP Authentication Scheme
We can get its decoded value by using request.authorization .
{'username': 'user@user.com', 'password': 'Passw0rd'}Flask decodes the Authorization header for us. We see the Flask decode code in the following block.
# Flask authorization header parser
def parse_authorization_header(
value: t.Optional[str],
) -> t.Optional["ds.Authorization"]:
"""Parse an HTTP basic/digest authorization header transmitted by the web
browser. The return value is either `None` if the header was invalid or
not given, otherwise an :class:`~werkzeug.datastructures.Authorization`
object.
:param value: the authorization header to parse.
:return: a :class:`~werkzeug.datastructures.Authorization` object or `None`.
"""
if not value:
return None
value = _wsgi_decoding_dance(value)
try:
auth_type, auth_info = value.split(None, 1)
auth_type = auth_type.lower()
except ValueError:
return None
if auth_type == "basic": # We used this scheme by sending request with httpx
try:
username, password = base64.b64decode(auth_info).split(b":", 1) # httpx encodes the header for us
except Exception:
return None
try:
return ds.Authorization(
"basic",
{
"username": _to_str(username, "utf-8"),
"password": _to_str(password, "utf-8"),
},
)
except UnicodeDecodeError:
return None
elif auth_type == "digest":
auth_map = parse_dict_header(auth_info)
for key in "username", "realm", "nonce", "uri", "response":
if key not in auth_map:
return None
if "qop" in auth_map:
if not auth_map.get("nc") or not auth_map.get("cnonce"):
return None
return ds.Authorization("digest", auth_map)
return NoneAnd after checking if the user is exists in the system, we can create JWT by using our createJWT function. We can extend as what we want the payload.
adminkey of the payload is created to handle “authorization” which is not in our scope for this text.
def createJWT(username, secret, authz):
payload = {
"username": username,
"exp": datetime.datetime.now(tz=datetime.timezone.utc) # expire date
+ datetime.timedelta(days=1), # token lifetime is 1 day
"iat": datetime.datetime.utcnow(), # issued at
"admin": authz, # is user admin
}
return jwt.encode(
payload,
secret,
algorithm="HS256",
) # return JWT tokenIf we make a successfull request to our login endpoint, we’ll get the JWT token.
Now we are ready to validate the token by using our validate endpoint.
@app.route("/validate", methods=["POST"])
def validate():
encoded_jwt = request.headers[
"Authorization"
] # handle "Authorization" header. We have to send the token in this scheme => Bearer <TOKEN>
if not encoded_jwt:
return ("missing credentials", 401)
encoded_jwt = encoded_jwt.split(" ")[1] # get the token (exclude "Bearer")
try:
decoded = jwt.decode(
encoded_jwt, JWT_SECRET, algorithms=["HS256"]
) # decode the token
except:
return ("not authorized", 403)
return (decoded, 200)If we can make a successfull request to our validate endpoint, we need to decoded JWT.
httpx -m POST -h "Authorization" "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6InVzZXJAdXNlci5jb20iLCJleHAiOjE2NzA4NDY4NjIsImlhdCI6MTY3MDc2MDQ2MiwiYWRtaW4iOnRydWV9.Z3bA45Q414VsO9GYECsr9YO40ac1xD-EuMz5TeUOroA" http://127.0.0.1:5000/validate
We’ve successfully developed our auth service. Now we need to create our post service. Please keep in mind, we’ll use this validate and login endpoint from the post service to authenticate the user.
Gateway Service
We’ll start by defining the constants.
"""
# We use the snippets below to make requests
# login post request
- httpx -m POST --auth user@user.com Passw0rd http://127.0.0.1:8080/login
# GET Posts
- httpx -m GET -h "Authorization" "Bearer <TOKEN>" http://127.0.0.1:8080/posts
# POST Posts
- httpx -m POST -h "Authorization" "Bearer <TOKEN>" -j '{"id":1, "title": "Title Text"}' http://127.0.0.1:8080/posts
"""
from flask import Flask, request
import httpx
import json
AUTH_SERVICE_URL = "http://127.0.0.1:5000" # you should get this by using ENV
POSTS_DB = [] # to simulate inserting
app = Flask(__name__)We’ll use AUTH_SERVICE_URL for setting prefix in our making requests functions.
POSTS_DB is being our dummy database.
appis the server.
Actually, in here I tried to simulate API Gateway architecture. However, to keep it simple, I also combined the post service in the gateway service. Therefore, in my example there is just one extra service to simulate posts & gateway.
Firstly, I am going to start by explaining the gateway functions in the posts service.
def auth_service_login(request):
"""Auth request to auth service
Args:
request (request): Flask request
Returns:
tuple: First item of the tuple will be the token if the request is successfull and 2nd item will be the status code
"""
auth = request.authorization
if not auth:
# if there is no "Authorization" header in the request that is coming to "gateway" service, return error
return None, ("missing credentials", 401)
basicAuth = (auth.username, auth.password)
# post request to auth service to get the token
response = httpx.post(f"{AUTH_SERVICE_URL}/login", auth=basicAuth)
if response.status_code == 200:
return response.text, None
else:
return None, (response.text, response.status_code)This auth_service_login function helps us to be logged in by redirecting the post service request to auth service. If it is successfull, we will have a JWT and None for error. If it is not successfull, we will have None for token and error.
def auth_service_validate_token(request):
"""Token validation in auth service
Args:
request (request): Flask request
Returns:
tuple: First item of the tuple will be the decoded token if the request is successfull and 2nd item will be the status code
"""
if not "Authorization" in request.headers:
# if there is no "Authorization" header in the request that is coming to "gateway" service, return error
return None, ("missing credentials", 401)
token = request.headers["Authorization"]
if not token:
# if there is no JWT token return error
return None, ("missing credentials", 401)
# post request to auth service to validate the token
response = httpx.post(
f"{AUTH_SERVICE_URL}/validate",
headers={"Authorization": token},
)
# auth service returns us the token and status code
if response.status_code == 200:
return response.text, None
else:
return None, (response.text, response.status_code)This auth_service_validate_token function redirects post service request to auth service validate endpoint. It works in the same way with the auth_service_login function.
Now we are ready to implement endpoints of this service.
@app.route("/login", methods=["POST"])
def login():
# get the token by using gateway service (we can think that the request will be redirected to the auth service)
token, err = auth_service_login(request)
if not err:
return token
else:
return err
@app.route("/posts", methods=["POST"])
def post_posts():
# firstly validate the token by using gateway service. By doing that we enforce to make the user to be logged in
token, err = auth_service_validate_token(request)
if err:
return err
else:
POSTS_DB.append(request.json)
return {"msg": "success", "data": request.json}
@app.route("/posts", methods=["GET"])
def get_posts():
# firstly validate the token by using gateway service. By doing that we enforce to make the user to be logged in
token, err = auth_service_validate_token(request)
if err:
return err
token = json.loads(token) # we get decoded token
return {"posts": POSTS_DB}We are ready to experiment gateway service.
Gateway Testing
I am going to make a request to gateway service’s login endpoint.
I’ve separated the services by using different port numbers.
5000forauthservice and8080forgatewayservice.
httpx -m POST --auth user@user.com Passw0rd http://127.0.0.1:8080/login
Now we can get the posts list by using the token we got from gateway login endpoint.
httpx -m GET -h "Authorization" "Bearer <TOKEN>" http://127.0.0.1:8080/posts
We can add new posts.
httpx -m POST -h "Authorization" "Bearer <TOKEN>" -j '{"id":1, "title": "Title Text"}' http://127.0.0.1:8080/posts
httpx -m POST -h "Authorization" "Bearer <TOKEN>" -j '{"id":2, "title": "Post 2"}' http://127.0.0.1:8080/posts
httpx -m POST -h "Authorization" "Bearer <TOKEN>" -j '{"id":3, "title": "Post 3"}' http://127.0.0.1:8080/posts
If everything went good, we should see the posts we added.
httpx -m GET -h "Authorization" "Bearer <TOKEN>" http://127.0.0.1:8080/posts
Finally
Hopefully, you enjoyed and found helpful. I wanted to enhance my knowledge and learn the basics of authentication. I no longer just copy and paste the code to make it run (to be honest, I already wasn’t doing that 🥲).
You can find the code we used in the text by following the link below.
Also, you can subscribe my Youtube channel to see the texts in action in video 🤓
Regards
