Add device authorization grant (device code flow - rfc 8628)

AUTHORS

@@ -42,6 +42,7 @@ Dave Burkholder
 David Fischer
 David Hill
 David Smith
+David Uzumaki
 Dawid Wolski
 Diego Garcia
 Dominik George

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [unreleased]
 ### Added
 * #1506 Support for Wildcard Origin and Redirect URIs
+* Add device authorization grant support
 <!--
 ### Changed
 ### Deprecated

docs/tutorial/tutorial.rst

@@ -9,4 +9,4 @@ Tutorials
    tutorial_03
    tutorial_04
    tutorial_05
-
+   tutorial_06

docs/tutorial/tutorial_06.rst

@@ -0,0 +1,100 @@
+Device authorization grant flow
+====================================================
+
+Scenario
+--------
+In :doc:`Part 1 <tutorial_01>` you created your own :term:`Authorization Server` and it's running along just fine.
+You have devices that your users have, and those users need to authenticate the device against your
+:term:`Authorization Server` in order to make the required API calls.
+
+Device Authorization
+--------------------
+The OAuth 2.0 device authorization grant is designed for Internet
+connected devices that either lack a browser to perform a user-agent
+based authorization or are input-constrained to the extent that
+requiring the user to input text in order to authenticate during the
+authorization flow is impractical. It enables OAuth clients on such
+devices (like smart TVs, media consoles, digital picture frames, and
+printers) to obtain user authorization to access protected resources
+by using a user agent on a separate device.
+
+Point your browser to `http://127.0.0.1:8000/o/applications/register/` to create an application.
+
+Fill the form as shown in the screenshot below, and before saving, take note of the ``Client id``.
+Make sure the client type is set to "Public." There are cases where a confidential client makes sense,
+but generally, it is assumed the device is unable to safely store the client secret.
+
+.. image:: _images/application-register-device-code.png
+   :alt: Device Authorization application registration
+
+Ensure the setting ``OAUTH_DEVICE_VERIFICATION_URI`` is set to a URI you want to return in the
+`verification_uri` key in the response. This is what the device will display to the user.
+
+1: Navigate to the tests/app/idp directory:
+
+.. code-block:: sh
+
+    cd tests/app/idp
+
+then start the server
+.. code-block:: sh
+
+    python manage.py runserver
+
+To initiate device authorization, send this request:
+
+.. code-block:: sh
+
+    curl --location 'http://127.0.0.1:8000/o/device-authorization/' \
+        --header 'Content-Type: application/x-www-form-urlencoded' \
+        --data-urlencode 'client_id={your application\'s client id}'
+
+The OAuth2 provider will return the following response:
+
+.. code-block:: json
+
+    {
+        "verification_uri": "http://127.0.0.1:8000/o/device",
+        "expires_in": 1800,
+        "user_code": "A32RVADM",
+        "device_code": "G30j94v0kNfipD4KmGLTWeL4eZnKHm",
+        "interval": 5
+    }
+
+Go to `http://127.0.0.1:8000/o/device` in your browser.
+
+.. image:: _images/device-enter-code-displayed.png
+
+Enter the code, and it will redirect you to the device-confirm endpoint.
+
+Device-confirm endpoint
+-----------------------
+Device polling occurs concurrently while the user approves or denies the request.
+
+.. image:: _images/device-approve-deny.png
+
+Device polling
+--------------
+Note: You should already have the `/token` endpoint implemented in your authorization server before this.
+
+Send the following request (in the real world, the device makes this request):
+
+.. code-block:: sh
+
+    curl --location 'http://localhost:8000/o/token/' \
+        --header 'Content-Type: application/x-www-form-urlencoded' \
+        --data-urlencode 'device_code={the device code from the device-authorization response}' \
+        --data-urlencode 'client_id={your application\'s client id}' \
+        --data-urlencode 'grant_type=urn:ietf:params:oauth:grant-type:device_code'
+
+The response will be similar to this:
+
+.. code-block:: json
+
+    {
+        "access_token": "SkJMgyL432P04nHDPyB63DEAM0nVxk",
+        "expires_in": 36000,
+        "token_type": "Bearer",
+        "scope": "openid",
+        "refresh_token": "Go6VumurDfFAeCeKrpCKPDtElV77id"
+    }

oauth2_provider/migrations/0013_alter_application_authorization_grant_type_device.py

@@ -0,0 +1,41 @@
+# Generated by Django 5.1.5 on 2025-01-24 14:00
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('oauth2_provider', '0012_add_token_checksum'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='application',
+            name='authorization_grant_type',
+            field=models.CharField(choices=[('authorization-code', 'Authorization code'), ('urn:ietf:params:oauth:grant-type:device_code', 'Device Code'), ('implicit', 'Implicit'), ('password', 'Resource owner password-based'), ('client-credentials', 'Client credentials'), ('openid-hybrid', 'OpenID connect hybrid')], max_length=44),
+        ),
+        migrations.CreateModel(
+            name='Device',
+            fields=[
+                ('id', models.BigAutoField(primary_key=True, serialize=False)),
+                ('device_code', models.CharField(max_length=100, unique=True)),
+                ('user_code', models.CharField(max_length=100)),
+                ('scope', models.CharField(max_length=64, null=True)),
+                ('interval', models.IntegerField(default=5)),
+                ('expires', models.DateTimeField()),
+                ('status', models.CharField(blank=True, choices=[('authorized', 'Authorized'), ('authorization-pending', 'Authorization pending'), ('expired', 'Expired'), ('denied', 'Denied')], default='authorization-pending', max_length=64)),
+                ('client_id', models.CharField(db_index=True, max_length=100)),
+                ('last_checked', models.DateTimeField(auto_now=True)),
+                ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='%(app_label)s_%(class)s', to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+                'swappable': 'OAUTH2_PROVIDER_DEVICE_MODEL',
+                'constraints': [models.UniqueConstraint(fields=('device_code',), name='oauth2_provider_device_unique_device_code')],
+            },
+        ),
+    ]

oauth2_provider/models.py

@@ -3,7 +3,10 @@
 import time
 import uuid
 from contextlib import suppress
-from datetime import timedelta
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from datetime import timezone as dt_timezone
+from typing import Callable, Optional, Union
 from urllib.parse import parse_qsl, urlparse
 
 from django.apps import apps
@@ -86,12 +89,14 @@ class AbstractApplication(models.Model):
     )
 
     GRANT_AUTHORIZATION_CODE = "authorization-code"
+    GRANT_DEVICE_CODE = "urn:ietf:params:oauth:grant-type:device_code"
     GRANT_IMPLICIT = "implicit"
     GRANT_PASSWORD = "password"
     GRANT_CLIENT_CREDENTIALS = "client-credentials"
     GRANT_OPENID_HYBRID = "openid-hybrid"
     GRANT_TYPES = (
         (GRANT_AUTHORIZATION_CODE, _("Authorization code")),
+        (GRANT_DEVICE_CODE, _("Device Code")),
         (GRANT_IMPLICIT, _("Implicit")),
         (GRANT_PASSWORD, _("Resource owner password-based")),
         (GRANT_CLIENT_CREDENTIALS, _("Client credentials")),
@@ -127,7 +132,7 @@ class AbstractApplication(models.Model):
         default="",
     )
     client_type = models.CharField(max_length=32, choices=CLIENT_TYPES)
-    authorization_grant_type = models.CharField(max_length=32, choices=GRANT_TYPES)
+    authorization_grant_type = models.CharField(max_length=44, choices=GRANT_TYPES)
     client_secret = ClientSecretField(
         max_length=255,
         blank=True,
@@ -650,11 +655,110 @@ class Meta(AbstractIDToken.Meta):
         swappable = "OAUTH2_PROVIDER_ID_TOKEN_MODEL"
 
 
+class AbstractDevice(models.Model):
+    class Meta:
+        abstract = True
+        constraints = [
+            models.UniqueConstraint(
+                fields=["device_code"],
+                name="%(app_label)s_%(class)s_unique_device_code",
+            ),
+        ]
+
+    AUTHORIZED = "authorized"
+    AUTHORIZATION_PENDING = "authorization-pending"
+    EXPIRED = "expired"
+    DENIED = "denied"
+
+    DEVICE_FLOW_STATUS = (
+        (AUTHORIZED, _("Authorized")),
+        (AUTHORIZATION_PENDING, _("Authorization pending")),
+        (EXPIRED, _("Expired")),
+        (DENIED, _("Denied")),
+    )
+
+    id = models.BigAutoField(primary_key=True)
+    user = models.ForeignKey(
+        settings.AUTH_USER_MODEL,
+        related_name="%(app_label)s_%(class)s",
+        null=True,
+        blank=True,
+        on_delete=models.CASCADE,
+    )
+    device_code = models.CharField(max_length=100, unique=True)
+    user_code = models.CharField(max_length=100)
+    scope = models.CharField(max_length=64, null=True)
+    interval = models.IntegerField(default=5)
+    expires = models.DateTimeField()
+    status = models.CharField(
+        max_length=64, blank=True, choices=DEVICE_FLOW_STATUS, default=AUTHORIZATION_PENDING
+    )
+    client_id = models.CharField(max_length=100, db_index=True)
+    last_checked = models.DateTimeField(auto_now=True)
+
+    def is_expired(self):
+        """
+        Check device flow session expiration.
+        """
+        now = datetime.now(tz=dt_timezone.utc)
+        return now >= self.expires
+
+
+class DeviceManager(models.Manager):
+    def get_by_natural_key(self, client_id, device_code, user_code):
+        return self.get(client_id=client_id, device_code=device_code, user_code=user_code)
+
+
+class Device(AbstractDevice):
+    objects = DeviceManager()
+
+    class Meta(AbstractDevice.Meta):
+        swappable = "OAUTH2_PROVIDER_DEVICE_MODEL"
+
+    def natural_key(self):
+        return (self.client_id, self.device_code, self.user_code)
+
+
+@dataclass
+class DeviceRequest:
+    # https://datatracker.ietf.org/doc/html/rfc8628#section-3.1
+    # scope is optional
+    client_id: str
+    scope: Optional[str] = None
+
+
+@dataclass
+class DeviceCodeResponse:
+    verification_uri: str
+    expires_in: int
+    user_code: int
+    device_code: str
+    interval: int
+    verification_uri_complete: Optional[Union[str, Callable]] = None
+
+
+def create_device(device_request: DeviceRequest, device_response: DeviceCodeResponse) -> Device:
+    now = datetime.now(tz=dt_timezone.utc)
+
+    return Device.objects.create(
+        client_id=device_request.client_id,
+        device_code=device_response.device_code,
+        user_code=device_response.user_code,
+        scope=device_request.scope,
+        expires=now + timedelta(seconds=device_response.expires_in),
+    )
+
+
 def get_application_model():
     """Return the Application model that is active in this project."""
     return apps.get_model(oauth2_settings.APPLICATION_MODEL)
 
 
+def get_device_model():
+    """Return the Device model that is active in this project."""
+    return apps.get_model(oauth2_settings.DEVICE_MODEL)
+
+
 def get_grant_model():
     """Return the Grant model that is active in this project."""
     return apps.get_model(oauth2_settings.GRANT_MODEL)

oauth2_provider/oauth2_backends.py

@@ -1,6 +1,7 @@
 import json
 from urllib.parse import urlparse, urlunparse
 
+from django.http import HttpRequest
 from oauthlib import oauth2
 from oauthlib.common import Request as OauthlibRequest
 from oauthlib.common import quote, urlencode, urlencoded
@@ -75,6 +76,8 @@ def extract_headers(self, request):
             del headers["wsgi.errors"]
         if "HTTP_AUTHORIZATION" in headers:
             headers["Authorization"] = headers["HTTP_AUTHORIZATION"]
+        if "CONTENT_TYPE" in headers:
+            headers["Content-Type"] = headers["CONTENT_TYPE"]
         # Add Access-Control-Allow-Origin header to the token endpoint response for authentication code grant,
         # if the origin is allowed by RequestValidator.is_origin_allowed.
         # https://github.com/oauthlib/oauthlib/pull/791
@@ -148,6 +151,16 @@ def create_authorization_response(self, request, scopes, credentials, allow):
         except oauth2.OAuth2Error as error:
             raise OAuthToolkitError(error=error, redirect_uri=credentials["redirect_uri"])
 
+    def create_device_authorization_response(self, request: HttpRequest):
+        uri, http_method, body, headers = self._extract_params(request)
+        try:
+            headers, body, status = self.server.create_device_authorization_response(
+                uri, http_method, body, headers
+            )
+            return headers, body, status
+        except OAuth2Error as exc:
+            return exc.headers, exc.json, exc.status_code
+
     def create_token_response(self, request):
         """
         A wrapper method that calls create_token_response on `server_class` instance.

oauth2_provider/oauth2_validators.py

@@ -52,10 +52,12 @@
     "client_credentials": (AbstractApplication.GRANT_CLIENT_CREDENTIALS,),
     "refresh_token": (
         AbstractApplication.GRANT_AUTHORIZATION_CODE,
+        AbstractApplication.GRANT_DEVICE_CODE,
         AbstractApplication.GRANT_PASSWORD,
         AbstractApplication.GRANT_CLIENT_CREDENTIALS,
         AbstractApplication.GRANT_OPENID_HYBRID,
     ),
+    "urn:ietf:params:oauth:grant-type:device_code": (AbstractApplication.GRANT_DEVICE_CODE,),
 }
 
 Application = get_application_model()
@@ -166,6 +168,11 @@ def _authenticate_basic_auth(self, request):
         elif request.client.client_id != client_id:
             log.debug("Failed basic auth: wrong client id %s" % client_id)
             return False
+        elif (
+            request.client.client_type == "public"
+            and request.grant_type == "urn:ietf:params:oauth:grant-type:device_code"
+        ):
+            return True
         elif not self._check_secret(client_secret, request.client.client_secret):
             log.debug("Failed basic auth: wrong client secret %s" % client_secret)
             return False
@@ -191,6 +198,11 @@ def _authenticate_request_body(self, request):
         if self._load_application(client_id, request) is None:
             log.debug("Failed body auth: Application %s does not exists" % client_id)
             return False
+        elif (
+            request.client.client_type == "public"
+            and request.grant_type == "urn:ietf:params:oauth:grant-type:device_code"
+        ):
+            return True
         elif not self._check_secret(client_secret, request.client.client_secret):
             log.debug("Failed body auth: wrong client secret %s" % client_secret)
             return False