Demystifying Device Flow
Implementing OAuth 2.0 Device Authorization Grant with AWS Cognito and FastAPI
Have you ever wondered how the Netflix app on your smart TV knows which account is linked to it? Or why you should scan a QR code when the TV requires you to link to your app? In this blogpost we will dive deep into the Device Authorization Grant protocol, which is exactly what Netflix uses. We will implement this protocol ourselves from scratch.
In this blogpost we will implement OAuth Device Authorization Grant using AWS Cognito and FastAPI. At the end, you will be able to log into your app from devices with limited browser accessibility.
We implemented this flow for authenticating a Golang CLI that interacts with the Data Product Portal, our brand new open-source project. You can find more information in the announcement blogpost or on our Github!
What is Device Authorization Grant?
The OAuth 2.0 Device Authorization Grant, is a standardized way of authorizing headless devices. Headless devices are devices with no web browsers and/or limited capability for inputting text. Authorizing with Device Authorization Grant is meant for use on devices that don’t have an easy way of entering text, such as a TV, as it stops you from having to input passwords on that device. Famous examples of Device Authorization Grant include
- logging into the AWS CLI (headless device) through SSO
- connecting your smart TV (headless device) to your Netflix account
- connecting your printer (headless device) to your printer OEM account
In Device Authorization Grant, the headless device shows a QR code or a URL where the user can then complete authentication using a secondary device (like a smartphone).
In this blog post, we are going to implement a FastAPI backend for handling Device Authorization Grant with AWS Cognito.
Two flows
A schematic representation of the full Device Authorization Grant can be found below.
The Device Authorization Grant consists of two separate parts or flows that run partially in parallel: 1) The headless device flow and 2) The browser flow.
Next to that, four different actors all play a role in the Device Authorization Grant.
Different actors
Before we dive into the two flows, let’s briefly discuss all the actors that play a part.
- User: The actor that initiates the authentication request on the headless device and performs the actual authentication on a secondary device.
- Headless Device: Device that wants to be authenticated, e.g. smart TV or CLI.
- Cognito: authorization server
- Backend: authorization client, an API hosting the endpoints for Device Authorization Grant. This is what we will implement during this blogpost.
The headless device flow
The “Headless Device”, such as your smart TV or CLI, will prompt you with a user code and a verification URL. We will call this the first flow, it originates from the headless device, after you make clear you want to authenticate the device, e.g. by pressing the Login button on your smart TV, or by typing the commandaws sso login.
The browser flow
The second flow is initiated by you, as the acting user, when you navigate to the verification URL, provided by the headless device flow. This can be done in a variety of ways: Netflix shows you a QR code to scan with your phone, CLIs can automatically open your browser, or you could simply retype the URL in your browser. When you load the URL you are first requested to confirm that the user code shown on the headless device matches, or you should input the provided user code yourself, and then you are required to authenticate to the API that you try to use. This authentication will in our case be handled by AWS Cognito, our OAuth identity provider.
Continuous polling by the headless device flow
While the user is trying to authenticate in a browser, the headless device flow continues in parallel. The headless device polls the authorization client — this is the backend we will implement — to check if the authentication succeeded and when that is the case it requests the actual id, access (and optionally refresh) token from the identity provider. The backend stores different verification codes to ensure the identity provider that the original requester of the authorization codes now fetches the tokens. This polling should start immediately after the user code, device code and verification uri are created and should be done periodically (typically 5 seconds delay or less). Polling frequently enough will result in the headless device knowing quickly when it is correctly authenticated and ensures a smooth user experience.
Upon successful retrieval of the access tokens, the Device Authorization Grant is concluded. The headless device can now access your API securely by authorizing requests with a JWT Bearer token.
The full documentation and explanation of Device Authorization Grant can be found in the following RFC.
Why implement Device Authorization Grant yourself?
Our authorization server, AWS Cognito, as well as many other authorization servers, unfortunately do not support a flow for headless devices. So in this blogpost, we will implement this ourselves. As there is already a standardized, well-thought out API for this called the Device Authorization Grant, we use this as inspiration (on top of the authorization code grant, and even though we are not implementing an authorization server). This requires (temporary) state.
Some OAuth clients have Device Authorization Grant ready to go with the endpoints they provide. However, if you decided on using AWS Cognito as your authorization authority, you are out of luck.
The only mention that AWS gives about Device Authorization Grant is the following blog post, which uses JavaScript and AWS Lambda. Sure, it is a starting point, but it might not fit in your current architecture or technology stack.
Our application and technology stack
We have developed “Data Product Portal”, a data engineering tool with a web UI frontend and a FastAPI-based backend. Our tool will also be accessible through a Go CLI. This CLI is the reason we need Device Authorization Grant.
The provided code samples will assume you have a working demo application running with FastAPI, including a small database component. You can follow the excellent documentation to achieve this or look at my Github for a working example.
This blog post does not cover the setup of AWS Cognito or the setup of AWS Cognito with FastAPI. We will only look at implementing Device Authorization Grant given a working AWS Cognito setup.
Implementation
Let’s dive into the real work. We will focus only on the relevant parts related to Device Authorization Grant. We will look specifically at implementing the Backend actor.
A full working code example, including all of the basic setups, can be found here.
Cognito setup
We assume your application already has a working AWS Cognito setup. Basic instructions can be found here. Make sure to integrate your FastAPI application as an app in your user pool. In the code below we will reference this configuration like this.
from app.core.auth.oidc import oidcoidc is a model that has the following attributes, filled in by Cognito.
client_id
client_secret
authority
redirect_uri
configuration_url # The .well-known/openid-configuration url
# These can be fetched from configuration_url
authorization_endpoint
token_endpoint
userinfo_endpoint
jwks_keysProject structure
We will create four files, following the structure FastAPI recommends when working in bigger projects:
model.pyThis file reflects the Pydantic model in our database. We will not include this code in the blogpost. You can find an example in my Github Reposchema.pyThis file contains the Pydantic model used by FastAPI to return responses or get request parameters. The Pydantic model is fairly simple. We will not include this code in the blogpost. You can find an example in my Github Reporouter.pyThis file defines the available API callsservice.pyThis file implements the functionality behind these calls
Requesting a device token on the headless device (Headless device flow)
This call will be made from the headless device to our backend. We request a new device token. This call will generate a device_code, unique to our device, a user_code to verify in the browser and a url to navigate to for authentication. This looks like this in router.py and service.py . Code shown here is pseudo-code and only shows some relevant concepts, please refer to the Github repo for the actual full implementation.
# router.py
router = APIRouter(prefix="/device") # include this in your main router
@router.post("/device_token")
async def get_device_token(
client_id: str,
# During device flow we authenticate with Basic HTTP Auth to pass client ID and secret
auth_client_id: Annotated[str, Depends(verify_auth_header)],
scope: str = "openid",
db: Session = Depends(get_db_session),
):
return DeviceFlowService().get_device_token(auth_client_id, client_id, db, scope)# service.py
class DeviceFlowService:
def generate_device_flow_codes(
self, db: Session, client_id: str, scope: str = "openid"
) -> DeviceFlow:
# Generate a new DeviceFlowModel
# This generates the device_code, user_code for a scope
# It saves the new devicecodes to the db
device_flow = DeviceFlowModel(
client_id=client_id,
scope=scope,
max_expiry=utc_now() + timedelta(seconds=1800),
oidc_redirect_uri=oidc.redirect_uri,
last_checked=utc_now(),
)
db.add(device_flow)
db.commit()
return DeviceFlow.model_validate(device_flow)Browser-Backend Communication (Browser flow)
Here we implement four possible calls.
- Landing page for confirming / denying a user code
- The confirm call
- The deny call
- The callback after the user is authenticated
# router.py
# Base url for /device route. The landing page.
# Show User code and allow confirm / deny by user.
@router.get("", include_in_schema=False)
async def request_user_code_processing(
code: str, request: Request, db: Session = Depends(get_db_session)
):
return DeviceFlowService().request_user_code_processing(code, request, db)
# User code does not match or request was unintended. Deny the device flow
@router.get("/deny", include_in_schema=False)
...
# User code matches and we want to authorize.
@router.get("/allow", include_in_schema=False)
...
# Callback for after the authentication in the browser completes
@router.get("/callback", include_in_schema=False)
...# service.py
def request_user_code_processing(
self, user_code: str, request: Request, db: Session
):
# Get all device flows with matching user code
device_flows = db.get(DeviceFlow, user_code)
# Make sure there is only one
...
device_flow = device_flows[0]
# Only valid state is PENDING
if device_flow.status in (
DeviceFlowStatus.EXPIRED,
DeviceFlowStatus.AUTHORIZED,
DeviceFlowStatus.DENIED,
):
raise ValueError("The device code has already been expired or been used")
# There is an expiry on device code requests.
if utc_now() > device_flow.max_expiry:
device_flow.status = DeviceFlowStatus.EXPIRED
db.commit()
raise ValueError("User Code has expired")
# User code is valid
# Generate a nice screen with allow / deny buttons
...
# Out of scope for this blogpost. This renders a jinja template to present the 2 buttons
return HTMLResponse(...)
# What happens when user presses deny
def deny_device_flow(self, device_code: str, db: Session):
device = db.get(DeviceFlowModel, device_code)
device.status = DeviceFlowStatus.DENIED
db.commit()
return RedirectResponse("/")
# When we allow the device flow.
# First generate some code challenges,
# these get passed to Cognito to ensure no tampering
# We also generate a state for the device request
# Afterward we redirect to Cognito for authentication.
# The redirect URI after cognito authentication links back to our callback
def allow_device_flow(self, client_id: str, device_code: str, db: Session):
code_verifier = uuid4().hex
hash = sha256(code_verifier.encode("utf-8")).digest()
code_challenge = urlsafe_b64encode(hash).decode("utf-8").replace("=", "")
state = uuid4().hex
device = db.get(DeviceFlowModel, device_code)
device.authz_state = state
# Code verifier is used later to check we are the original requesters for the token.
device.authz_verif = code_verifier
db.commit()
return RedirectResponse(
status_code=302,
url=(
f"{oidc.authorization_endpoint}?"
f"response_type=code&client_id={client_id}"
f"&scope={device.scope}&"
f"redirect_uri={oidc.redirect_uri}api/auth/device/callback/"
f"&state={state}&scope={device.scope}&code_challenge_method=S256"
f"&code_challenge={code_challenge}"
f"&identity_provider={oidc.provider.name}"
),
)
# Callback after cognito authentication.
# Check that Cognito returned state equals saved device request state
def process_authz_code_callback(self, authz_code: str, state: str, db: Session):
devices = (
db.query(DeviceFlowModel).filter(DeviceFlowModel.authz_state == state).all()
)
if len(devices) != 1:
raise NonUniqueDevicecodeException()
device = devices[0]
device.authz_code = authz_code
# As of now we can fetch the token back from the headless device
device.status = DeviceFlowStatus.AUTHORIZED
db.commit()
# Some more Jinja content, out of scope.
return HTMLResponse(...)Fetching the JWT token (Headless device flow)
The final call can be made from the headless device and runs in parallel with the browser-backend communication flow. After the initial request for a device code, this call will be made intermittently to the backend until it succeeds. See diagram above.
# router.py
@router.post("/jwt_token")
async def get_jwt_token(
request: Request,
client_id: str,
device_code: str,
grant_type: str,
auth_client_id: Annotated[str, Depends(verify_auth_header)],
db: Session = Depends(get_db_session),
):
return DeviceFlowService().get_jwt_token(
request, auth_client_id, client_id, device_code, grant_type, db
)# service.py
def get_jwt_token(
self,
request: Request,
auth_client_id: str,
client_id: str,
device_code: str,
grant_type: str,
db: Session,
):
# This is the only grant type allowed for device code
if grant_type != "urn:ietf:params:oauth:grant-type:device_code":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="POST Call on /jwt_token invalid: incorrect grant type",
)
assert self._verify_auth_header(auth_client_id, client_id)
return self.fetch_jwt_tokens(request, db, device_code, client_id)
def fetch_jwt_tokens(
self, request: Request, db: Session, device_code: str, client_id: str
):
# Make sure there only exists a single, still valid device code request
device_flow = db.get(DeviceFlowModel, device_code)
... # Same checks on validity and expiry of device flow
if utc_now() <= device_flow.last_checked + timedelta(
seconds=device_flow.interval
):
# We should not DOS our API for device code requests.
device_flow.last_checked = utc_now()
db.commit()
raise ValueError(f"Client makes too much API calls {utc_now()},\
{device_flow.last_checked}")
# This is the happy path. We still have a valid device code
# and we are on time for checking the status.
device_flow.last_checked = utc_now()
db.commit()
if device_flow.status in (
DeviceFlowStatus.AUTHORIZATION_PENDING,
DeviceFlowStatus.DENIED,
) or (
device_flow.status == DeviceFlowStatus.AUTHORIZED
and not device_flow.authz_code
):
# The user has not yet authenticated, or the request has been denied.
# This exception needs to be handled correctly by the headless device.
# Either check again in some seconds or break off the authorization process.
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail=device_flow.status
)
elif (
device_flow.status == DeviceFlowStatus.AUTHORIZED and device_flow.authz_code
):
# Happy flow, the user has correctly authenticated and all the codes match
# Make a request to the cognito token endpoint.
# With this request the code and verifier will be checked.
# This makes sure that we are the original requesters
response = httpx.post(
oidc.token_endpoint,
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": request.headers.get("Authorization"),
},
data={
"grant_type": "authorization_code",
"client_id": client_id,
"redirect_uri": f"{oidc.redirect_uri}api/auth/device/callback/",
"code": device_flow.authz_code,
"code_verifier": device_flow.authz_verif,
},
)
# Result will be access and (optionally) refresh tokens with the scope provided in the beginning.
data = response.json()
self.logger.debug(data)
# Expire so we can't use this device request again
device_flow.status = DeviceFlowStatus.EXPIRED
db.commit()
return data
else:
raise ValueError()That’s it! Congratulations! You have now implemented a working OAuth Device Authorization Grant. Enjoy the ability to log in from headless devices to your API.
Conclusion
For a full implementation of Device Authorization Grant, visit my Github at https://github.com/stijn-janssens/cognito-fastapi-device-flow
Let’s briefly summarize: First of all we talked about the what, why and how of Device Authorization Grant. Secondly you have learned to implement Device Authorization Grant on your backend. In order to do that you need to implement the browser flow, the headless device flow and store authorization codes in between. You can style the look and feel of the different landing pages as you prefer. Feel free to try it out for yourself!
Make sure to refer to the official OAuth 2.0 and Cognito documentation for more detailed information and further customization options.
A production-grade implementation of this Device Authorization Grant can be found in our open-source Data Product Portal. Feel free to fork our Github. Device authorization grant can be found here.
Thanks for reading all the way to the end :)
