Entmystifizierung des Geräteflusses

Haben Sie sich jemals gefragt, wie die Netflix-App auf Ihrem Smart-TV weiß, welches Konto damit verknüpft ist? Oder warum Sie einen QR-Code scannen sollten, wenn der Fernseher von Ihnen verlangt, sich mit Ihrer App zu verknüpfen? In diesem Blogbeitrag werden wir tief in das Protokoll zur Geräteautorisierung eintauchen, das genau das ist, was Netflix verwendet. Wir werden dieses Protokoll selbst von Grund auf implementieren.

In diesem Blogbeitrag werden wir die OAuth-Geräteautorisierung mit AWS Cognito und FastAPI implementieren. Am Ende werden Sie in der Lage sein, sich von Geräten mit eingeschränkter Browserzugänglichkeit in Ihre App einzuloggen.

Wir haben diesen Ablauf für die Authentifizierung eines Golang CLI implementiert, das mit dem Data Product Portal interagiert, unserem brandneuen Open-Source-Projekt. Weitere Informationen finden Sie im Ankündigungs-Blogbeitrag oder auf unserem Github!

Was ist die Geräteautorisierung?

Die OAuth 2.0 Geräteautorisierung ist eine standardisierte Methode zur Autorisierung von headless Geräten. Headless Geräte sind Geräte ohne Webbrowser und/oder mit eingeschränkten Möglichkeiten zur Texteingabe. Die Autorisierung mit der Geräteautorisierung ist für die Verwendung auf Geräten gedacht, die keine einfache Möglichkeit haben, Text einzugeben, wie z.B. ein Fernseher, da es Sie daran hindert, Passwörter auf diesem Gerät eingeben zu müssen. Bekannte Beispiele für die Geräteautorisierung sind

• das Einloggen in die AWS CLI (headless Gerät) über SSO

• das Verbinden Ihres Smart TVs (headless Gerät) mit Ihrem Netflix-Konto

• das Verbinden Ihres Druckers (headless Gerät) mit Ihrem Drucker-OEM-Konto

Bei der Geräteautorisierung zeigt das headless Gerät einen QR-Code oder eine URL, über die der Benutzer die Authentifizierung mit einem sekundären Gerät (wie einem Smartphone) abschließen kann.

In diesem Blogbeitrag werden wir ein FastAPI-Backend zur Handhabung der Geräteautorisierung mit AWS Cognito implementieren.

Zwei Abläufe

Eine schematische Darstellung der vollständigen Geräteautorisierung finden Sie unten.

Die Geräteautorisierung besteht aus zwei separaten Teilen oder Abläufen, die teilweise parallel laufen: 1) DerHeadless-Geräte-Ablauf und 2) DerBrowser-Ablauf.

Zusätzlich spielen vier verschiedene Akteure eine Rolle bei der Geräteautorisierung.

Verschiedene Akteure

Bevor wir in die beiden Abläufe eintauchen, lassen Sie uns kurz alle Akteure besprechen, die eine Rolle spielen.

• Benutzer: Der Akteur, der die Authentifizierungsanforderung auf dem headless Gerät initiiert und die tatsächliche Authentifizierung auf einem sekundären Gerät durchführt.

• Headless Gerät: Gerät, das authentifiziert werden möchte, z.B. Smart TV oder CLI.

• Cognito: Autorisierungsserver

• Backend: Autorisierungsklient, ein API, das die Endpunkte für die Geräteautorisierung hostet. Dies werden wir während dieses Blogbeitrags implementieren.

  
  

  
  

Der Headless-Geräte-Ablauf

Das „Headless-Gerät“, wie Ihr Smart-TV oder CLI, wird Ihnen einen Benutzer-Code und eine Verifikations-URL anzeigen. Wir nennen dies den ersten Ablauf, der vom headless Gerät ausgeht, nachdem Sie klargestellt haben, dass Sie das Gerät authentifizieren möchten, z.B. durch Drücken der Anmelden-Taste auf Ihrem Smart-TV oder durch Eingabe des Befehlsaws sso login.

Der Browser Ablauf

Der zweite Ablauf wird von Ihnen initiiert, als agierender Benutzer, wenn Sie zur Verifikations-URL navigieren, die vom Headless-Geräte-Ablauf bereitgestellt wird. Dies kann auf verschiedene Weise geschehen: Netflix zeigt Ihnen einen QR-Code, den Sie mit Ihrem Telefon scannen können, CLIs können automatisch Ihren Browser öffnen, oder Sie könnten die URL einfach erneut in Ihrem Browser eingeben. Wenn Sie die URL laden, werden Sie zuerst gebeten zu bestätigen, dass der Benutzer-Code, der auf dem headless Gerät angezeigt wird, übereinstimmt, oder dass Sie den bereitgestellten Benutzer-Code selbst eingeben, und dann müssen Sie sich bei der API authentifizieren, die Sie verwenden möchten. Diese Authentifizierung wird in unserem Fall von AWS Cognito, unserem OAuth-Identitätsanbieter, behandelt.

Kontinuierliches Polling durch den Headless-Geräte-Ablauf

Während der Benutzer versucht, sich in einem Browser zu authentifizieren, läuft der Headless-Geräte-Ablauf parallel weiter. Das headless Gerät pollt den Autorisierungsklienten – das ist das Backend, das wir implementieren werden – um zu überprüfen, ob die Authentifizierung erfolgreich war, und wenn dies der Fall ist, fordert es das tatsächliche ID-, Zugriffs- (und optional Refresh-) Token vom Identitätsanbieter an. Das Backend speichert verschiedene Verifizierungscodes, um dem Identitätsanbieter zu versichern, dass der ursprüngliche Anforderer der Autorisierungscodes jetzt die Token abruft. Dieses Polling sollte sofort beginnen, nachdem der Benutzer-Code, der Geräte-Code und die Verifikations-URI erstellt wurden, und sollte regelmäßig (typischerweise alle 5 Sekunden oder weniger) erfolgen. Häufiges Polling führt dazu, dass das headless Gerät schnell erfährt, wann es korrekt authentifiziert ist und sorgt für ein reibungsloses Benutzererlebnis.

Nach erfolgreichem Abrufen der Zugriffstoken wird die Geräteautorisierung abgeschlossen. Das headless Gerät kann nun sicher auf Ihre API zugreifen, indem es Anfragen mit einem JWT-Bearer-Token autorisiert.

Die vollständige Dokumentation und Erklärung der Geräteautorisierung finden Sie in dem folgenden RFC.

Warum die Geräteautorisierung selbst implementieren?

Unser Autorisierungsserver, AWS Cognito sowie viele andere Autorisierungsserver unterstützen leider keinen Ablauf für headless Geräte. Daher werden wir dies in diesem Blogbeitrag selbst implementieren. Da es bereits eine standardisierte, gut durchdachte API dafür gibt, die Geräteautorisierung heißt, verwenden wir dies als Inspiration (neben dem Autorisierungscode-Ablauf, auch wenn wir keinen Autorisierungsserver implementieren). Dies erfordert (vorübergehenden) Status.

Einige OAuth-Clients haben die Geräteautorisierung bereit, um die Endpunkte bereitzustellen, die sie anbieten. Wenn Sie jedoch beschlossen haben, AWS Cognito als Ihre Autoritätsbehörde zu verwenden, stehen Sie leider vor Problemen.

Die einzige Erwähnung, die AWS zur Geräteautorisierung macht, ist der folgende Blogbeitrag, der JavaScript und AWS Lambda verwendet. Sicher, es ist ein Ausgangspunkt, aber es passt möglicherweise nicht in Ihre aktuelle Architektur oder Technologie-Stack.

Unsere Anwendung und Technologie-Stack

Wir haben „Data Product Portal“ entwickelt, ein Data Engineering-Tool mit einem Web-UI-Frontend und einem FastAPI-basierten Backend. Unser Tool wird auch über ein Go CLI zugänglich sein. Dieses CLI ist der Grund, warum wir die Geräteautorisierung benötigen. Die bereitgestellten Codebeispiele gehen davon aus, dass Sie eine funktionierende Demoanwendung haben, die mit FastAPI ausgeführt wird, einschließlich einer kleinen Datenbankkomponente.

Sie können die hervorragende Dokumentation folgen, um dies zu erreichen oder mein Github für ein funktionierendes Beispiel ansehen. Dieser Blogbeitrag behandelt nicht die Einrichtung von AWS Cognito oder die Einrichtung von AWS Cognito mit FastAPI. Wir werden uns nur mit der Implementierung der Geräteautorisierung befassen, vorausgesetzt, dass bereits ein funktionierendes AWS Cognito eingerichtet ist.

ImplementierungTauchen wir ein in die eigentliche Arbeit. Wir werden uns nur auf die relevanten Teile konzentrieren, die mit der Geräteautorisierung zusammenhängen. Wir werden uns speziell mit der Implementierung des Backend Akteurs befassen. Ein vollständiges Arbeitsbeispiel, einschließlich aller grundlegenden Setups, ist hier zu finden. Cognito-Einrichtung Wir nehmen an, dass Ihre Anwendung bereits eine funktionierende AWS Cognito-Einrichtung hat. Grundlegende Anweisungen finden Sie hier. Stellen Sie sicher, dass Sie Ihre FastAPI-Anwendung als App in Ihrem Benutzerpool integrieren. Im folgenden Code werden wir diese Konfiguration wie folgt referenzieren.

Implementierung

Lassen Sie uns in die eigentliche Arbeit eintauchen. Wir werden uns nur auf die relevanten Teile konzentrieren, die mit der Geräteautorisierung zusammenhängen. Wir werden uns speziell mit der Implementierung des Backend Akteurs befassen.

Ein vollständiges funktionsfähiges Codebeispiel, einschließlich aller grundlegenden Setups, kann hier gefunden werden.

Cognito-Einrichtung

Wir nehmen an, dass Ihre Anwendung bereits eine funktionierende AWS Cognito-Einrichtung hat. Grundlegende Anweisungen finden Siehier. Stellen Sie sicher, dass Sie Ihre FastAPI-Anwendung als App in Ihrem Benutzerpool integrieren. Im folgenden Code werden wir diese Konfiguration wie folgt referenzieren.

from app.core.auth.oidc import oidc

oidc ist ein Modell, das die folgenden Attribute hat, die von Cognito ausgefüllt werden.

client_id
client_secret
authority
redirect_uri
configuration_url # Die .well-known/openid-configuration URL

# Diese können von configuration_url abgerufen werden
authorization_endpoint
token_endpoint
userinfo_endpoint
jwks_keys

Projektstruktur

Wir werden vier Dateien erstellen, die der Struktur folgen, die FastAPI für größere Projekte empfiehlt:

• model.py Diese Datei spiegelt das Pydantic-Modell in unserer Datenbank wider. Wir werden diesen Code im Blogbeitrag nicht einfügen. Ein Beispiel finden Sie in meinem Github Repo.

• schema.py Diese Datei enthält das Pydantic-Modell, das von FastAPI zur Rückgabe von Antworten oder zum Abrufen von Anfrageparametern verwendet wird. Das Pydantic-Modell ist recht einfach. Wir werden diesen Code im Blogbeitrag nicht einfügen. Ein Beispiel finden Sie in meinem Github Repo.

• router.py Diese Datei definiert die verfügbaren API-Aufrufe.

• service.py Diese Datei implementiert die Funktionalität hinter diesen Aufrufen.

Anfordern eines Gerätetokens auf dem headless Gerät (Headless-Geräte-Ablauf)

Dieser Aufruf wird vom headless Gerät an unser Backend gesendet. Wir fordern ein neues Gerätetoken an. Dieser Aufruf wird einen device_code generieren, der einzigartig für unser Gerät ist, einen user_code, um ihn im Browser zu verifizieren, und eine url, um zur Authentifizierung zu navigieren. So sieht es in router.py und service.py aus. Der hier gezeigte Code ist Pseudocode und zeigt nur einige relevante Konzepte. Bitte beziehen Sie sich auf das Github-Repo für die tatsächliche vollständige Implementierung.

# 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-Kommunikation (Browser-Ablauf)

Hier implementieren wir vier mögliche Aufrufe.

• Landing-Page zur Bestätigung/Verweigerung eines Benutzer-Codes

• Der Bestätigungsaufruf

• Der Verweigerungsaufruf

• Der Callback, nachdem sich der Benutzer authentifiziert hat

# router.py

# Basis-URL für die /device-Route. Die Landing-Page.
# Zeigt Benutzer-Code an und erlaubt Bestätigung/Verweigerung durch den Benutzer.
@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)

# Benutzer-Code stimmt nicht überein oder die Anfrage war unbeabsichtigt. Verweigern Sie den Geräteablauf
@router.get("/deny", include_in_schema=False)
...

# Benutzer-Code stimmt überein und wir wollen autorisieren.
@router.get("/allow", include_in_schema=False)
...

# Callback, nachdem die Authentifizierung im Browser abgeschlossen ist
@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(...)

JWT-Token abrufen (Headless-Geräte-Ablauf)

Der finale Aufruf kann vom headless Gerät getätigt werden und läuft parallel zum Browser-Backend-Kommunikationsablauf. Nach der ersten Anfrage für einen Gerätecode wird dieser Aufruf intermittierend an das Backend gesendet, bis er erfolgreich ist. Siehe Diagramm oben.

# 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()

Das war's! Herzlichen Glückwunsch! Sie haben jetzt eine funktionierende OAuth-Geräteautorisierung implementiert. Genießen Sie die Möglichkeit, sich von headless Geräten in Ihre API einzuloggen.

Fazit

Für eine vollständige Implementierung der Geräteautorisierung besuchen Sie mein Github unter https://github.com/stijn-janssens/cognito-fastapi-device-flow.

Zusammenfassend lassen Sie uns kurz zusammenfassen: Zunächst haben wir über das Was, Warum und Wie der Geräteautorisierung gesprochen. Zweitens haben Sie gelernt, wie Sie die Geräteautorisierung in Ihrem Backend implementieren können. Um dies zu tun, müssen Sie den Browser-Ablauf, den Headless-Geräte-Ablauf implementieren und Autorisierungscodes dazwischen speichern. Sie können das Aussehen und die Gestaltung der verschiedenen Landing-Pages nach Ihrem Geschmack gestalten. Fühlen Sie sich frei, es selbst auszuprobieren!

Stellen Sie sicher, dass Sie auf dieoffizielle OAuth 2.0undCognito-Dokumentationverweisen, um detailliertere Informationen und weitere Anpassungsoptionen zu erhalten.

Eine produktionsfähige Implementierung dieser Geräteautorisierung finden Sie in unserem Open-Source Data Product Portal. Fühlen Sie sich frei, unsere Github zu forken. Die Geräteautorisierung finden Sie hier.

Danke, dass Sie bis zum Ende gelesen haben :)