1
\$\begingroup\$

Context

I've been working on a Django-based Google Authentication application, designed to manage OAuth authentication without relying on libraries such as django-allauth. This is primarily intended for a backend that interfaces with a React frontend through API requests. My main goal is to establish authentication using oauthlib and Django's inherent capabilities.

Code Functionality:

The application is responsible for handling user authentication via Google, leveraging the OAuth protocol. Once authenticated, it should provide secure access to the React frontend via JWT tokens.

Main Concerns:

  • Authentication Best Practices: Does the authentication flow adheres to standard best practices, particularly with OAuth and Django?
  • Security Best Practices: I'm particularly interested in how the code handles security aspects such as CSRF, secure API communication, and protection against common vulnerabilities.
  • OOP Practices: Feedback on how well Object-Oriented Programming principles are integrated in regards to Django best-practices.
  • Code Separation of Concerns: How well the code separates different functionalities, such as handling OAuth logic, user specific action, token specific actions, etc. I've seen there's a Requests-OAuthlib but I'm not sure if it has any benefits.

Code

urls.py

from django.urls import path

from authentication.views import GoogleLoginApi, GoogleLoginRedirectApi

auth_urlpatterns = [
    path("auth/google/redirect/", GoogleLoginRedirectApi.as_view(), name="google-login-redirect"),
    path("auth/google/login/", GoogleLoginApi.as_view(), name="google-login"),
]

views.py

import hmac

from django.shortcuts import redirect
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_simplejwt.tokens import RefreshToken

from authentication.google_service import GoogleService
from authentication.serializers import InputSerializer
from users.models import CustomUser


class GoogleLoginRedirectApi(APIView):
    permission_classes = [AllowAny]

    def get(self, *args, **kwargs):
        google_service = GoogleService.get_service()
        authorization_url, state = google_service.get_authorization_url()
        self.request.session.update({"google_oauth2_state": state, "modified": True})
        return redirect(authorization_url)


class GoogleLoginApi(APIView):
    permission_classes = [AllowAny]

    def get(self, *args, **kwargs):
        input_serializer = InputSerializer(data=self.request.GET)
        input_serializer.is_valid(raise_exception=True)

        validated_data = input_serializer.validated_data

        code = validated_data["code"]
        state = validated_data["state"]

        error = validated_data.get("error")
        if error is not None:
            return Response({"error": error}, status=status.HTTP_400_BAD_REQUEST)

        session_state = self.request.session.get("google_oauth2_state")
        if session_state is None or not hmac.compare_digest(session_state, state):
            return Response({"error": "CSRF check failed."}, status=status.HTTP_400_BAD_REQUEST)

        del self.request.session["google_oauth2_state"]

        if state != session_state:
            return Response({"error": "CSRF check failed."}, status=status.HTTP_400_BAD_REQUEST)

        google_service = GoogleService.get_service()
        google_tokens = google_service.get_tokens(code=code)
        id_token_decoded = google_service.decode_id_token(google_tokens=google_tokens)

        user_info = google_service.get_user_info(google_tokens=google_tokens)
        google_id, email = id_token_decoded.get("sub"), id_token_decoded.get("email")

        user = CustomUser.get_or_create_from_google(
            google_id=google_id, email=email, user_info=user_info
        )

        refresh = RefreshToken.for_user(user)

        return Response(
            {
                "access_token": str(refresh.access_token),
                "refresh_token": str(refresh),
                "user_info": user_info,
            }
        )

serializers.py

from rest_framework import serializers


class InputSerializer(serializers.Serializer):
    code = serializers.CharField(required=True)
    error = serializers.CharField(required=False)
    state = serializers.CharField(required=True)

google_service.py

from random import SystemRandom
from urllib.parse import urlencode

import jwt
import requests
from django.conf import settings
from oauthlib.common import UNICODE_ASCII_CHARACTER_SET

from common.exceptions import ApplicationError


class GoogleService:
    _google_service = None

    def __init__(self):
        self._client_id = settings.GOOGLE_OAUTH2_CLIENT_ID
        self._client_secret = settings.GOOGLE_OAUTH2_CLIENT_SECRET

    @classmethod
    def get_service(cls):
        if cls._google_service is None:
            cls._google_service = GoogleService()
        return cls._google_service

    def get_authorization_url(self, length=30, chars=UNICODE_ASCII_CHARACTER_SET):
        rand = SystemRandom()
        state = "".join(rand.choice(chars) for _ in range(length))

        query_params = urlencode(
            {
                "response_type": "code",
                "client_id": self._client_id,
                "redirect_uri": settings.GOOGLE_AUTH_REDIRECT_URI,
                "scope": " ".join(settings.GOOGLE_AUTH_SCOPES),
                "state": state,
                "access_type": "offline",
                "include_granted_scopes": "true",
                "prompt": "select_account",
            }
        )
        authorization_url = f"{settings.GOOGLE_AUTH_URL}?{query_params}"

        return authorization_url, state

    def get_tokens(self, code: str) -> dict:
        response = requests.post(
            settings.GOOGLE_ACCESS_TOKEN_OBTAIN_URL,
            data={
                "code": code,
                "client_id": self._client_id,
                "client_secret": self._client_secret,
                "redirect_uri": settings.GOOGLE_AUTH_REDIRECT_URI,
                "grant_type": "authorization_code",
            },
        )
        if not response.ok:
            raise ApplicationError("Failed to obtain access token from Google.")

        google_tokens = response.json()

        return dict(
            access_token=google_tokens["access_token"],
            expires_in=google_tokens["expires_in"],
            refresh_token=google_tokens["refresh_token"],
            token_type=google_tokens["token_type"],
            id_token=google_tokens["id_token"],
        )

    def get_user_info(self, google_tokens: dict):
        response = requests.get(
            settings.GOOGLE_USER_INFO_URL, params={"access_token": google_tokens["access_token"]}
        )
        if not response.ok:
            raise ApplicationError("Failed to obtain user info from Google.")
        return response.json()

    def decode_id_token(self, google_tokens: dict) -> dict[str, str]:
        jwks_client = jwt.PyJWKClient(settings.GOOGLE_AUTH_CERTS_URI)
        signing_key = jwks_client.get_signing_ey_from_jwt(google_tokens["id_token"])

        return jwt.decode(
            jwt=google_tokens["id_token"],
            key=signing_key.key,
            algorithms=["RS256"],
            audience=settings.GOOGLE_OAUTH2_CLIENT_ID,
        )

users\models.py

from django.contrib.auth.models import AbstractUser
from django.db import models
from django.utils.translation import gettext_lazy as _

from users.managers import CustomUserManager


class CustomUser(AbstractUser):
    username = None
    email = models.EmailField(_("email address"), unique=True)

    # Fields used to store OAuth Provider "sub" ids
    google_id = models.CharField(max_length=255, unique=True, null=True, blank=True)

    USERNAME_FIELD = "email"
    REQUIRED_FIELDS = []

    objects = CustomUserManager()

    def __str__(self):
        return self.email

    @classmethod
    def get_or_create_from_google(cls, google_id, email, user_info):
        user = cls.objects.filter(google_id=google_id).first()
        if not user:
            user = cls.objects.filter(email=email).first()

            if user:
                user.google_id = google_id
                user.save(update_fields=["google_id"])
            else:
                user = cls.objects.create_user(
                    email=email,
                    first_name=user_info.get("given_name", ""),
                    last_name=user_info.get("family_name", ""),
                    is_active=user_info.get("email_verified", False),
                )
        return user

users\managers.py

from django.contrib.auth.base_user import BaseUserManager
from django.utils.translation import gettext_lazy as _


class CustomUserManager(BaseUserManager):
    """
    Custom user model manager where email is the unique identifiers
    for authentication instead of usernames.
    """

    def create_user(self, email, password=None, **extra_fields):
        """
        Create and save a user with the given email and password.
        """
        if not email:
            raise ValueError(_("The email must be set"))

        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)

        if password:
            user.set_password(password)
        else:
            # TODO: this creates an account with an unusable password. We should instead
            #  raise an error and not create an account if no password is provided for traditional
            #  login
            user.set_unusable_password()

        user.save()
        return user

    def create_superuser(self, email, password, **extra_fields):
        """
        Create and save a SuperUser with the given email and password.
        """
        extra_fields.setdefault("is_staff", True)
        extra_fields.setdefault("is_superuser", True)
        extra_fields.setdefault("is_active", True)

        if extra_fields.get("is_staff") is not True:
            raise ValueError(_("Superuser must have is_staff=True."))

        if extra_fields.get("is_superuser") is not True:
            raise ValueError(_("Superuser must have is_superuser=True."))

        return self.create_user(email, password, **extra_fields)

    def filter_by_google_id(self, google_id):
        return self.get_queryset().filter(google_id=google_id).first()

    def filter_by_email(self, email):
        return self.get_queryset().filter(email=email).first()

.env

SECRET_KEY=
DEBUG=True
ALLOWED_HOSTS=localhost,127.0.0.1

DB_ENGINE=django.db.backends.postgresql
DB_NAME=my_app
DB_USER=postgres
DB_PASSWORD=password
DB_HOST=db
DB_PORT=5432

# Google Auth

GOOGLE_OAUTH2_CLIENT_ID=
GOOGLE_OAUTH2_CLIENT_SECRET=
GOOGLE_OAUTH2_PROJECT_ID=

Other considerations:

The authentication flow should be fairly common and straight-forward:

  1. If the user is new, create a new record within CustomUser, link google_id (sub) and create new JWT access / refresh token.
  2. If the user is not new (record already exists in db), but they used other authentication method and now they use Google, link google_id and create new JWT access / refresh token.

Specific Questions:

  • Have I implemented the OAuth flow correctly and securely?
  • Are there any glaring security concerns in the way I've handled user data and authentication tokens?
  • How could the code structure be improved for better readability and maintenance?
  • Any suggestions for better aligning the code with Django's design philosophy and OOP principles?
  • I'm open to any constructive feedback and suggestions for improvement. Detailed explanations would be greatly appreciated.

Thank you in advance for your time and insights!

\$\endgroup\$
1
  • \$\begingroup\$ If someone's looking to test this locally, they'll have to obtain OAuth 2.0 client credentials from the Google API Console. \$\endgroup\$ Apr 2 at 20:41

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.