Securing API with AWS Cognito

Introduction

Couple weeks back I published an article: Deploying Complete Serverless API with any containerised Python ASGI Framework on AWS Lambda, API Gateway

I figured I needed to add a logical next step to securing the app. I was also partly inspired by the course by Steven Marek towards AWS Solutions Architect Certification, which I'm studiously following. Most importantly so you won’t turn out to be these guys:

Credit: DOGE.gov/unknown hacker

/src_code

The GitHub code repo for this project is here: https://github.com/652-Animashaun/blog-serverless

Its written in Python - FastAPI, deployed to AWS Lambda, I 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. But 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 interaction, 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. It’s a standard RS256 or HS256 base64encoding. And also add features like expiry. A refresh token.

A JWT contains 3 parts header, payload and a signature.

Payload is where the user identity like email, username are returned. Cognito returns 2 tokens, the access_token and the id_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!.

Dive right into the AWS Cognito dashboard create a user pool in 3 steps:

Creating and configuring a user pool:

1. Select Create user pool from the User pools menu, or select Get started for free in less than five minutes.

2. 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.

3. In Name your application, enter a descriptive name or proceed with the default name.

4. 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.

5. 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. 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": "muiz_shaun@gmail.com" } }

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 its done, the main thing is it is possible. And I like to preempt that, I want to do a quick verify.

/app.utils.auth.py

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), HTTPBearer will 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 or None in which case user will be redirected to /login.

Oh, and AWS Cognito is free for 10K monthly active users. Check it out:
Free Tier

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.

1. 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.

Oh, one more thing, remember, when you push changes to lambda function, add the api gateway endpoint version of your /token authorize() endpoint 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.

Read more:

https://docs.aws.amazon.com/cognito/latest/developerguide/getting-started-user-pools-application.html

https://pyjwt.readthedocs.io/en/stable/installation.html