Storing OAuth credentials in Django models

A common requirement for my Django apps is to interact with third-party APIs on behalf of its users, most of which use OAuth for authentication. I now realise Python Social Auth is a solution but I wasn’t aware of it at the time. In any case, here’s my approach in case the aforementioned package doesn’t fit your needs.

My apps use Requests OAuthlib which provides a Requests Session object that does the OAuth magic behind the scenes.

Models #

BaseOAuthModel #

This will store the access & refresh tokens as well as their metadata (expiry, etc), and have the static parameters such as client ID, client secret, on it as attributes.

class BaseOAuthModel(models.Model):
    class Meta:
        abstract = True

    refresh_token = models.TextField()
    access_token = models.TextField()
    expires_at = models.DateTimeField()
    token_type = models.TextField()

    objects = BaseOAuthModelManager()

    _session = None

    def update_token(self, token):
        self.access_token = token['access_token']
        self.refresh_token = token['refresh_token']
        self.expires_at = timezone.now() + timedelta(seconds=token['expires_in'])
        self.token_type = token['token_type']

        return self.save(update_fields=[
            'access_token', 'refresh_token', 'expires_at', 'token_type'
        ], force_update=True)

    def to_dict(self) -> dict:
        return {
            'access_token': self.access_token,
            'refresh_token': self.refresh_token,
            'expires_in': math.floor((self.expires_at - timezone.now()).total_seconds()),
            'token_type': self.token_type,
        }

    @property
    def session(self) -> OAuth2Session:
        if not self._session:
            extra = {
                'client_id': self.client_id,
                'client_secret': self.client_secret
            }
            self._session = OAuth2Session(client_id=self.client_id,
                                          token=self.to_dict(),
                                          auto_refresh_url=self.refresh_url,
                                          auto_refresh_kwargs=extra,
                                          token_updater=self.update_token)
        return self._session

update_token is called by Requests OAuthlib when the token was renewed - this updates the model and saves the new token in the database, ready for next use. This is the part that took me the most effort as I couldn’t really figure out a clean way of doing this - this is the best I came up with so far.

The model has a session property that gives you an OAuth2Session with the model’s credentials - you can use it right away to make requests or pass it to something like Uplink or popget.

BaseOAuthModelManager #

class BaseOAuthModelManager(models.Manager):
    def get_login_url(self, request) -> str:
        redirect_uri = request.build_absolute_uri(reverse(
            f'admin:{self.model._meta.app_label}_{self.model._meta.model_name}_oauth_callback'
        ))
        url, state = OAuth2Session(self.model.client_id, redirect_uri=redirect_uri). \
            authorization_url(self.model.auth_url)

        return url

    def create_from_oauth_callback(self, request) -> models.Model:
        redirect_uri = request.build_absolute_uri(reverse(
            f'admin:{self.model._meta.app_label}_{self.model._meta.model_name}_oauth_callback'
        ))
        authorization_code = request.GET.get('code')

        session = OAuth2Session(self.model.client_id, redirect_uri=redirect_uri)

        token = session.fetch_token(
            self.model.refresh_url,
            code=authorization_code,
            client_id=self.model.client_id,
            client_secret=self.model.client_secret
        )

        expires_at = timezone.now() + timedelta(seconds=token['expires_in'])

        return self.model(
            access_token=token['access_token'],
            refresh_token=token['refresh_token'],
            expires_at=expires_at,
            token_type=token['token_type']
        )

This manager allows us to create the models based on OAuth callbacks from the remote server. It also provides a way to generate a login URL which redirects back to an endpoint in the admin (since this is for internal use, but you are free to implement something similar in your user-facing views).

Logging in from Django Admin #

Another requirement is to allow logging into OAuth accounts from the Django Admin (the app I’m using this in only provides an API and doesn’t have any user-facing view besides the admin).

We can easily override the “add” view to instead redirect to the authentication URL (provided by the model manager), and provide an endpoint for the callback to actually create the model once the user provides consent.

class BaseOAuthModelAdmin(admin.ModelAdmin):
    """A ModelAdmin that supports creating OAuth client models."""

    def add_view(self, request, form_url='', extra_context=None) -> HttpResponseRedirect:
        """Redirect to the OAuth authentication URI."""
        if not self.has_add_permission(request):
            raise PermissionDenied

        return redirect(self.model.objects.get_login_url(request))

    def get_urls(self) -> list:
        """Add the OAuth callback URL to the admin URLs."""

        def wrap(view):
            def wrapper(*args, **kwargs):
                return self.admin_site.admin_view(view)(*args, **kwargs)

            wrapper.model_admin = self
            return update_wrapper(wrapper, view)

        return super().get_urls() + [
            url(r'oauth_callback', wrap(self.process_oauth_callback),
                name=f'{self.model._meta.app_label}_{self.model._meta.model_name}_oauth_callback')
        ]

    def process_oauth_callback(self, request) -> HttpResponse:
        """Create new object based on OAuth callback."""
        if not self.has_add_permission(request):
            raise PermissionDenied

        obj = self.model.objects.create_from_oauth_callback(request)
        self.save_model(request, obj, None, False)

        change_message = [{'added': {}}]
        self.log_addition(request, obj, change_message)

        return super().response_add(request, obj)

How to use #

To interact with a service you’ll need to subclass the above abstract model as follows (using Monzo as an example):

class MonzoAccount(BaseOAuthModel):
    class Meta:
        verbose_name = 'Monzo Account'

    client_id = settings.MONZO_CLIENT_ID
    client_secret = settings.MONZO_CLIENT_SECRET

    auth_url = settings.MONZO_AUTH_URL
    base_url = settings.MONZO_BASE_URL
    refresh_url = urljoin(base_url, '/oauth2/token')

    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='monzo_accounts')

You’ll need to define your static parameters like the client ID, secret and authentication URLs on the model itself.

If you need to do extra processing before adding an object (like associating it with the currently logged-in user), you can subclass the manager as well:

class MonzoAccountManager(BaseOAuthModelManager):
    def create_from_oauth_callback(self, request):
        obj = super().create_from_oauth_callback(request)

        obj.user = request.user  # Assign to current user

        return obj

In the manager, create_from_oauth_callback already returns you a model with a usable session so you can make requests in there right away, for example if you need to fetch some data from the newly connected account (screen name? Account ID on the remote service?) and save it on the model.

If using a custom manager don’t forget to explicitly define it on the model like objects = MonzoAccountManager().

Finally register your models with the Admin class as follows, in admin.py:

admin.site.register(MonzoAccount, BaseOAuthModelAdmin)

Enjoy!

 
1
Kudos
 
1
Kudos

Now read this

Setting an alternative shell in macOS Terminal

After switching to Zsh on my new Mac I noticed a little issue with the built-in Terminal. When using the default Bash shell, the app could tell whether a process (besides the shell) was running in it and present a confirmation if you... Continue →