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) Theheadless device flowand 2) Thebrowser 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.

ImplementationLet’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 setupWe 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.

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

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 foundhere. 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 oidc

oidc 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_keys

Project structure

We will create four files, following the structure FastAPI recommends when working in bigger projects:

• model.py This 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 Repo

• schema.py This 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 Repo

• router.py This file defines the available API calls

• service.py This 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.pyrouter = 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.pyclass 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.pydef 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 theofficial OAuth 2.0andCognito documentationfor 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 :)