Introduction
I published an article on hash node a couple weeks ago titled: [Deploying Complete Serverless API with any containerised Python ASGI Framework on AWS Lambda, API][1]
I figured I needed to add a logical next step to securing the api. I was also partly inspired by the course by Steven Marek towards AWS Solutions Architect Certification, which I'm studying.
/src_code
The GitHub code repo for this project is here: https://github.com/652-Animashaun/blog-serverless.git
Its written in Python - FastAPI, deployed to AWS Lambda, then put an API Gateway in front to route client requests to appropriate endpoints. That's it you're all caught up.
While code example used here is Fast API framework, I implore you the implementation discussed is python framework agnostic. Come on I'll show you.
What is AWS Cognito?
I must admit the AWS Cognito documentation gave me the run around. I spent way more time than I thought I should implementing this mechanism. I'm not all that new to JWT authentication. I've implemented mechanisms using the OAuth before.
Anyways I’ve done the work, gone through the belly of the beast that is the AWS documentation, so that you won’t have to. I’ll try to cut out the fluff.
Amazon Cognito is an identity platform for web and mobile apps. It’s a
user directory, an authentication server, and an authorization service
for OAuth 2.0 access tokens and AWS credentials. With Amazon Cognito,
you can authenticate and authorize users from the built-in user
directory, from your enterprise directory, and from consumer identity
providers like Google and Facebook.
So off the bat, we can already see several use cases for the AWS Cognito service. For now we want a simple authentication process, i.e user is served a login page to enter credentials, redirected to homepage on successful login, or presented a register page to create an account get OTP via provided email to verify ID. They get a token (JWT) which is presented to the server’s gate keeper for some resource access, in this case an API endpoint.
What is happening under the hood is simple enough too. Well relatively, if you compare to implementing the mechanism from scratch.
User authentication with Cognito user pools
Login & Register Pages - Cognito
While you can build and customise to your liking, but AWS Cognito comes with a ready-to-use login and register pages where you can just redirect users to, you’ll also provide a redirect url where the returned token can be processed to authorize user.
Sending & Verifying OTP - Cognito
Usually one would be inclined to use a service like SES or SNS on AWS or Twilio, but you’ll have to manually call or implement the mechanism from the server side. AWS cognito takes care of this out of the box.
Encoding & Signing Tokens - Cognito
Cognito also takes care of encoding tokens. RS256 or HS256 base64encoding, also add features like expiry, a refresh token if you want. A Json Web Token contains 3 parts header, payload and a signature. Payload is where the user identity like email, username are returned. Cognito returns 3 tokens, the access_token, id_token and a refresh_token, more about that later.
Decoding and Granting Access - Server
When a token is returned to the server, the receiver of the JWT verifies the signature using the secret key or the public key. And maybe go a step further and include it in request header so that subsequent requests can bear the authorisation token.
Look, point is Cognito takes a lot of manual implementation out of the developers hand. I’m sold!.
Let's dive right into the AWS Cognito dashboard create a user pool in 3 steps.
Creating and configuring a user pool
Select Create user pool from the User pools menu, or select Get started for free in less than five minutes.
Under Define your application, I’m choosing “traditional web application” you can choose the Application type that best fits the application scenario that you want to create authentication and authorization services for.
In Name your application, enter a descriptive name or proceed with the default name.
You must make some basic choices under Configure options that support settings that you can't change after you create your user pool. For this use case I’m selecting username, email and full name.
Add a return url on the server that handles payload receipts and process the payload. in this case mine is http://localhost:8000/api/v1/users/token for development and testing purposes. You can add more redirect url options. Once you're ready to update lambda function with local changes, we'll be sure to add another return url , this time with the API Gateway stage endpoint. Else we get some funny error client_id redirect_uri mismatch. Hit create.
Web App Server
Lets take a look in the user pool thats just been created:
Under recommendations you should go ahead and click “setup your web app”.
AWS was kind enough to show some code examples for different languages.
Selecting python will give you a quick preview how things would look like:
As shown, the framework in Cognitgo example is Flask, but it doesnt matter, that about how we’ll be implementing ours. I can already see that to start with I’ll need a /login endpoints and an /authorize endpoints at a minimum.
/app.api.enpoints.users.py
import os
import jwt
from pprint import pprint
from fastapi import APIRouter, HTTPException, status, Depends
from starlette.requests import Request
from app.utils.auth import oauth, verify_token
from starlette.responses import RedirectResponse
router = APIRouter()
@router.get("/login")
async def login(request: Request):
return await oauth.oidc.authorize_redirect(request, "http://localhost:8000/api/v1/users/token")
@router.get("/token")
async def authorize(request: Request):
payload = await oauth.oidc.authorize_access_token(request)
token = payload["id_token"]
res = await verify_token(token)
user = payload["userinfo"]
return token
These endpoints aren’t doing much heavy lifting. /login simply calls on oidc.authorize_redirect() provided by the oauth import from utils where OAuth was instanced and passed to oauth, more on that subsequently. It takes a request and a redirect url to process response.
async def authorize(request) will receive a get request on signing completion on cognito, with a json payload that looks like:
{ "id_token": "eyJraWQiOiJ3ekxUT1lrNVR....", "access_token": "eyJraWQiOiI3b0pqeFhY...", "refresh_token": "eyJjdHkiOiJKV1QiLC....", "expires_in": 3600, "token_type": "Bearer", "expires_at": 1738976151, "userinfo": { "at_hash": "aYWSJq...", "sub": "b4d8e428-f091-7040-80a5-f2", "email_verified": true, "iss": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_ftTN1aEr4", "cognito:username": "shaun", "nonce": "SYqZdIip5wBjjI10983", "origin_jti": "edc6ytb3c-4870-4979-8672-7d9a261ff471", "aud": "47re0a7e8l43ok719rtqcqnjih", "token_use": "id", "auth_time": 1738972551, "exp": 1738976151, "iat": 1738972551, "jti": "f20cf2a4-5c3f-44f9-8d0b-919816d41f38", "email": "*Emails are not allowed*" } }
As seen above we can already grab userinfo and quickly see the user ID. But theoretically some skilled man in the middle can provide their own ID, I dont know how it's done, the main thing is it is possible. And I like to preempt that, I want to do a quick verify.
import os
import jwt
from jwt import PyJWKClient
from authlib.integrations.starlette_client import OAuth
from fastapi import APIRouter, HTTPException, status, Depends, Security, HTTPException
from datetime import date, datetime, time, timedelta, timezone
from typing import Annotated
from app.models.users import User
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials, OAuth2PasswordBearer
from starlette.responses import RedirectResponse
from starlette.requests import Request
signin_url = 'https://cognito-idp.us-east-1.amazonaws.com/us-east-1_ftTN1aEr4/.well-known/jwks.json'
security = HTTPBearer(auto_error=False)
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
CLIENT_ID = os.environ["CLIENT_ID"]
ISSUER = "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_ftTN1aEr4"
oauth = OAuth()
oauth.register(
name='oidc',
authority=ISSUER,
client_id=CLIENT_ID,
client_secret=CLIENT_SECRET,
server_metadata_url='https://cognito-idp.us-east-1.amazonaws.com/us-east-1_ftTN1aEr4/.well-known/openid-configuration',
client_kwargs={'scope': 'email openid phone'}
)
async def verify_token(token:str)-> User:
jwks_client = PyJWKClient(signin_url)
try:
signing_key = jwks_client.get_signing_key_from_jwt(token)
payload = jwt.decode(
token,
signing_key,
audience=CLIENT_ID,
options={"verify_exp": True},
algorithms=["RS256"],
)
user = User(username=payload["cognito:username"], email=payload["email"], access_token=token)
return User
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token has expired")
except jwt.InvalidTokenError as e:
raise HTTPException(status_code=403, detail="Invalid token")
async def get_current_user(credentials: Annotated[HTTPAuthorizationCredentials, Security(security)]):
if not credentials:
return False
try:
return await verify_token(credentials.credentials)
except HTTPException:
return False
Lets do a quick run through the code:
The singing_url is provided by congnito in the userpool dashboard, this an endpoint provided by Cognito. At the back of it is a json_body of keys used to sign the token. More on that subsequently.
In FastApi using Oauth2 as dependency is through the Security() class provided. You can read more about the stuff here. TLDR fastapi.security.HTTPBearer is one of the tools provided to use called as a dependable during visit of protect endpoints. It’s use case is for bearer token authentication. auto_error=False because as per docs, by default, if the HTTP Bearer token is not provided (in an Authorization header), it'll automatically cancel the request and send the client an error. I don’t want that, I want to catch that and redirect to login so user can try challenge again.
HTTPBearer returns an object HTTPAuthorizationCredentials which I believe is received via the request header and provides it to the application.
Then we just shamelessly copied the boiler plate code on cognito recommendation page to instantiate Oauth.register().
async def verify_token() will take in a token, first it want’s to retrieve signing keys from the signing_url using the python module PyJWKClient class provided by JWT module. Next it calls jwt.decode(), takes the token, signing_key and a couple other keyword arguments. The decoded payload contains a dictionary with several user identifying information, but it’ll go ahead, returning username and email.
In async def get_current_user() where the first injection takes place. I’ve provided the HTTPAuthorizationCredentials as HTTPBearer security which would try to retrieve the automaticall from the incoming request header the Bearer Authorization this is where the token is expected to be for an authenticated request.
/app.api.endpoints.posts.py
To show the authentication mechanism actually works we have to create an endpoint that is protected:
from fastapi import APIRouter, Depends, HTTPException, status
from typing_extensions import Annotated
from app.utils.auth import oauth, verify_token, get_current_user
from app.models.users import User
from starlette.requests import Request
from starlette.responses import RedirectResponse
router = APIRouter()
@router.get("/")
async def all_posts(current_user: Annotated[User, Depends(get_current_user)]):
if not current_user:
return RedirectResponse(url="/api/v1/users/login")
print("current_user", current_user)
return {"Welcome": "Shaun's Blog", "user": current_user}
Pretty straightforward right? async def all_posts depends on get_current_user which in turns returns a user object or None, in which case user will be redirected to /login.
Oh did I mention and AWS Cognito is free for 10K monthly active users. Check it out:
Amazon Cognito Essentials and Lite have a free tier. The free tier
does not automatically expire at the end of your 12-month AWS Free
Tier term, and it is available to both existing and new AWS customers
indefinitely. Please note - the free tier pricing isn’t available in
the AWS GovCloud (US-West) region.
- For users who sign in directly via Amazon Cognito or through a social identity provider, Amazon Cognito user pools has a free tier of
10,000 monthly active user (MAU) per month per account or per AWS
organization. This free tier is applicable for customers that
configure their user pools to either the Lite or Essentials tier.
There is no free tier for the Plus tier.
Stick around next one up is integrating social ID providers like google social media accounts. Would be fun.
One more thing, remember when you push changes to lambda, add the api gateway endpoint version of your /token (authorise) to the list of return urls in the Cognito UserPool dashboard. This might be the cause of client_id redirect_uri mismatch error, watch out.
References:
https://docs.aws.amazon.com/cognito/latest/developerguide/getting-started-user-pools-application.html
https://pyjwt.readthedocs.io/en/stable/installation.html
