I am trying to learn and use AWS Cognito User Pools and integrate with and API implemented with Python FastAPI. So far I am using Authorisation Code Flow with my Cognito user pool redirecting to an endpoint on FastAPI to resolve the code challenge. The source code is appended at the end of this query.
The API has the following endpoints:
aws_cognito_redirect endpoint resolves the code challenge by sending the code challenge, redirect_uri, client_id etc. through to the AWS Cognito user pool oauth2/token endpoint. I can see in the console log output that the identity, access and refresh tokens were successfully retrieved.The FastAPI will additionally have some protected endpoints that will be called from a web application. Also there will be a web form to interact with the endpoints.
At this stage I could implement and host the webforms with FastAPI jinja2 templates. If I went with this option, presumably I could have the /aws_cognito_redirect endpoint return the tokens in a HTTP only session cookie. That way each subsequent client request would automatically include the cookie with no tokens exposed in the browser local storage. I am aware that I would have to deal with XSRF/CSRF with this option.
Alternatively I could implement the front end using Angular/React. Presumably, the recommended practice appears to be that I would have to reconfigure the authorisation flow to be Auth Code with PKCE? In that case, the Angular/React web client would then be communicating with AWS Cognito directly, to retrieve tokens that would be forwarded to the FastAPI endpoints. These tokens would be stored in the browser's local storage and then sent in an Authorisation Header for each subsequent request. I am aware that this approach is subject to XSS attacks.
Of the two, given my requirements, I think that I am leaning towards hosting the webapp on FastAPI using jinja2 templates and returning a HTTP Only session cookie on successfull sign in.
If I chose this implementation route, is there a FastAPI feature or Python library that allows an endpoint to be decorated/marked with auth required that would inspect the presence of the session cookie and perform token verification?
FastAPI
import base64
from functools import lru_cache
import httpx
from fastapi import Depends, FastAPI, Request
from fastapi.responses import RedirectResponse
from . import config
app = FastAPI()
@lru_cache()
def get_settings():
"""Create config settings instance encapsulating app config."""
return config.Settings()
def encode_auth_header(client_id: str, client_secret: str):
"""Encode client id and secret as base64 client_id:client_secret."""
secret = base64.b64encode(
bytes(client_id, "utf-8") + b":" + bytes(client_secret, "utf-8")
)
return "Basic " + secret.decode()
@app.get("/")
def read_root(settings: config.Settings = Depends(get_settings)):
login_url = (
"https://"
+ settings.domain
+ ".auth."
+ settings.region
+ ".amazoncognito.com/login?client_id="
+ settings.client_id
+ "&response_type=code&scope=email+openid&redirect_uri="
+ settings.redirect_uri
)
print("Redirecting to " + login_url)
return RedirectResponse(login_url)
@app.get("/aws_cognito_redirect")
async def read_code_challenge(
request: Request, settings: config.Settings = Depends(get_settings)
):
"""Retrieve tokens from oauth2/token endpoint"""
code = request.query_params["code"]
print("/aws_cognito_redirect received code := ", code)
auth_secret = encode_auth_header(settings.client_id, settings.client_secret)
headers = {"Authorization": auth_secret}
print("Authorization:" + str(headers["Authorization"]))
payload = {
"client_id": settings.client_id,
"code": code,
"grant_type": "authorization_code",
"redirect_uri": settings.redirect_uri,
}
token_url = (
"https://"
+ settings.domain
+ ".auth."
+ settings.region
+ ".amazoncognito.com/oauth2/token"
)
async with httpx.AsyncClient() as client:
tokens = await client.post(
token_url,
data=payload,
headers=headers,
)
print("Tokens\n" + str(tokens.json()))
FastAPI highly relies on the dependency injection, which can be used for the authentication too. All you need to do is to write a simple dependency that will check the cookie:
async def verify_access(secret_token: Optional[str] = Cookie(None)):
if secret_token is None or secret_token not in valid_tokens:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
)
return secret_token
And use it in your views as a dependency:
@app.get("/")
def read_root(settings: config.Settings = Depends(get_settings), auth_token = Depends(verify_access)):
...
If you want to secure a group of endpoints, you can define additional router that will always include verify_access as a dependency:
app = FastAPI()
auth_required_router = APIRouter()
app.include_router(
auth_required_router, dependencies=[Depends(verify_access)],
)
@auth_required_router.get("/")
def read_root(settings: config.Settings = Depends(get_settings)):
...
Note that the value returned by your authentication dependency is arbitrary, so you can return there anything that is meaningful in your use case (for example authenticated user account). If you want to retrieve this value in a view registered by the auth_required_router, simply define this dependency in your view parameters as well. FastAPI will resolve (and execute) this dependency only once.
You can even do something more complex, like creating 2 nested dependencies, one simply checking the authentication and the 2nd one retrieving the user account from the database:
async def authenticate(...):
... # Verifies the auth data without fetching the user
async def get_auth_user(auth = Depends(authenticate):
... # Gets the user from the database, based on the auth data
Now, your auth_required_router can have only the authenticate dependency, but every view that also needs access to the current user, can have additionally get_auth_user dependency defined, so the authentication will occur always (and always only once) and fetching the user from the database will occur only when needed.
You can learn more about security architecture in the FastAPI (and how to use the built-in support for the OAuth2) in the documentation