Blog example for Google OAuth2 with Django

.env.example

@@ -9,3 +9,7 @@ AWS_S3_SECRET_ACCESS_KEY=""
 AWS_STORAGE_BUCKET_NAME=""
 AWS_S3_REGION_NAME=""
 AWS_S3_CUSTOM_DOMAIN=""
+
+DJANGO_GOOGLE_OAUTH2_CLIENT_ID=""
+DJANGO_GOOGLE_OAUTH2_CLIENT_SECRET=""
+DJANGO_GOOGLE_OAUTH2_PROJECT_ID=""

.gitignore

@@ -134,3 +134,5 @@ dmypy.json
 
 node_modules/
 package*.json
+
+google_client_secret.json

config/django/base.py

@@ -27,7 +27,6 @@
 
 ALLOWED_HOSTS = ["*"]
 
-
 # Application definition
 
 LOCAL_APPS = [
@@ -99,7 +98,6 @@
 
 WSGI_APPLICATION = "config.wsgi.application"
 
-
 # Database
 # https://docs.djangoproject.com/en/3.0/ref/settings/#databases
 
@@ -120,7 +118,6 @@
         }
     }
 
-
 # Password validation
 # https://docs.djangoproject.com/en/3.0/ref/settings/#auth-password-validators
 
@@ -141,7 +138,6 @@
 
 AUTH_USER_MODEL = "users.BaseUser"
 
-
 # Internationalization
 # https://docs.djangoproject.com/en/3.0/topics/i18n/
 
@@ -155,7 +151,6 @@
 
 USE_TZ = True
 
-
 # Static files (CSS, JavaScript, Images)
 # https://docs.djangoproject.com/en/3.0/howto/static-files/
 
@@ -178,6 +173,7 @@
 from config.settings.cors import *  # noqa
 from config.settings.email_sending import *  # noqa
 from config.settings.files_and_storages import *  # noqa
+from config.settings.google_oauth2 import *  # noqa
 from config.settings.jwt import *  # noqa
 from config.settings.sentry import *  # noqa
 from config.settings.sessions import *  # noqa

config/settings/cors.py

@@ -1,2 +1,8 @@
+from config.env import env
+
 CORS_ALLOW_CREDENTIALS = True
 CORS_ALLOW_ALL_ORIGINS = True
+
+BASE_BACKEND_URL = env.str("DJANGO_BASE_BACKEND_URL", default="http://localhost:8000")
+BASE_FRONTEND_URL = env.str("DJANGO_BASE_FRONTEND_URL", default="http://localhost:3000")
+CORS_ORIGIN_WHITELIST = env.list("DJANGO_CORS_ORIGIN_WHITELIST", default=[BASE_FRONTEND_URL])

config/settings/google_oauth2.py

@@ -0,0 +1,5 @@
+from config.env import env
+
+GOOGLE_OAUTH2_CLIENT_ID = env.str("DJANGO_GOOGLE_OAUTH2_CLIENT_ID", default="")
+GOOGLE_OAUTH2_CLIENT_SECRET = env.str("DJANGO_GOOGLE_OAUTH2_CLIENT_SECRET", default="")
+GOOGLE_OAUTH2_PROJECT_ID = env.str("DJANGO_GOOGLE_OAUTH2_PROJECT_ID", default="")

mypy.ini

@@ -33,3 +33,11 @@ ignore_missing_imports = True
 [mypy-rest_framework_jwt.*]
 # Remove this when rest_framework_jwt stubs are present
 ignore_missing_imports = True
+
+[mypy-google_auth_oauthlib.*]
+# Remove this when google_auth_oauthlib stubs are present
+ignore_missing_imports = True
+
+[mypy-oauthlib.*]
+# Remove this when oauthlib stubs are present
+ignore_missing_imports = True

requirements/base.txt

@@ -21,3 +21,10 @@ attrs==22.1.0
 
 gunicorn==20.1.0
 sentry-sdk==1.14.0
+
+requests==2.30.0
+
+google-api-python-client==2.86.0
+google-auth==2.18.0
+google-auth-httplib2==0.1.0
+google-auth-oauthlib==1.0.0

styleguide_example/api/urls.py

@@ -5,4 +5,7 @@
     path("users/", include(("styleguide_example.users.urls", "users"))),
     path("errors/", include(("styleguide_example.errors.urls", "errors"))),
     path("files/", include(("styleguide_example.files.urls", "files"))),
+    path(
+        "google-oauth2/", include(("styleguide_example.blog_examples.google_login_server_flow.urls", "google-oauth2"))
+    ),
 ]

styleguide_example/blog_examples/google_login_server_flow/raw/apis.py

@@ -0,0 +1,82 @@
+from django.contrib.auth import login
+from django.shortcuts import redirect
+from rest_framework import serializers, status
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from styleguide_example.blog_examples.google_login_server_flow.raw.service import (
+    GoogleRawLoginFlowService,
+)
+from styleguide_example.users.selectors import user_list
+
+
+class PublicApi(APIView):
+    authentication_classes = ()
+    permission_classes = ()
+
+
+class GoogleLoginRedirectApi(PublicApi):
+    def get(self, request, *args, **kwargs):
+        google_login_flow = GoogleRawLoginFlowService()
+
+        authorization_url, state = google_login_flow.get_authorization_url()
+
+        request.session["google_oauth2_state"] = state
+
+        return redirect(authorization_url)
+
+
+class GoogleLoginApi(PublicApi):
+    class InputSerializer(serializers.Serializer):
+        code = serializers.CharField(required=False)
+        error = serializers.CharField(required=False)
+        state = serializers.CharField(required=False)
+
+    def get(self, request, *args, **kwargs):
+        input_serializer = self.InputSerializer(data=request.GET)
+        input_serializer.is_valid(raise_exception=True)
+
+        validated_data = input_serializer.validated_data
+
+        code = validated_data.get("code")
+        error = validated_data.get("error")
+        state = validated_data.get("state")
+
+        if error is not None:
+            return Response({"error": error}, status=status.HTTP_400_BAD_REQUEST)
+
+        if code is None or state is None:
+            return Response({"error": "Code and state are required."}, status=status.HTTP_400_BAD_REQUEST)
+
+        session_state = request.session.get("google_oauth2_state")
+
+        if session_state is None:
+            return Response({"error": "CSRF check failed."}, status=status.HTTP_400_BAD_REQUEST)
+
+        del request.session["google_oauth2_state"]
+
+        if state != session_state:
+            return Response({"error": "CSRF check failed."}, status=status.HTTP_400_BAD_REQUEST)
+
+        google_login_flow = GoogleRawLoginFlowService()
+
+        google_tokens = google_login_flow.get_tokens(code=code)
+
+        id_token_decoded = google_tokens.decode_id_token()
+        user_info = google_login_flow.get_user_info(google_tokens=google_tokens)
+
+        user_email = id_token_decoded["email"]
+        request_user_list = user_list(filters={"email": user_email})
+        user = request_user_list.get() if request_user_list else None
+
+        if user is None:
+            return Response({"error": f"User with email {user_email} is not found."}, status=status.HTTP_404_NOT_FOUND)
+
+        login(request, user)
+
+        result = {
+            "id_token_decoded": id_token_decoded,
+            "user_info": user_info,
+        }
+
+        return Response(result)

styleguide_example/blog_examples/google_login_server_flow/raw/service.py

@@ -0,0 +1,133 @@
+from random import SystemRandom
+from typing import Any, Dict
+from urllib.parse import urlencode
+
+import jwt
+import requests
+from attr import define
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+from django.urls import reverse_lazy
+from oauthlib.common import UNICODE_ASCII_CHARACTER_SET
+
+from styleguide_example.core.exceptions import ApplicationError
+
+
+@define
+class GoogleRawLoginCredentials:
+    client_id: str
+    client_secret: str
+    project_id: str
+
+
+@define
+class GoogleAccessTokens:
+    id_token: str
+    access_token: str
+
+    def decode_id_token(self) -> Dict[str, str]:
+        id_token = self.id_token
+        decoded_token = jwt.decode(jwt=id_token, options={"verify_signature": False})
+        return decoded_token
+
+
+class GoogleRawLoginFlowService:
+    API_URI = reverse_lazy("api:google-oauth2:login-raw:callback-raw")
+
+    GOOGLE_AUTH_URL = "https://accounts.google.com/o/oauth2/auth"
+    GOOGLE_ACCESS_TOKEN_OBTAIN_URL = "https://oauth2.googleapis.com/token"
+    GOOGLE_USER_INFO_URL = "https://www.googleapis.com/oauth2/v3/userinfo"
+
+    SCOPES = [
+        "https://www.googleapis.com/auth/userinfo.email",
+        "https://www.googleapis.com/auth/userinfo.profile",
+        "openid",
+    ]
+
+    def __init__(self):
+        self._credentials = google_raw_login_get_credentials()
+
+    @staticmethod
+    def _generate_state_session_token(length=30, chars=UNICODE_ASCII_CHARACTER_SET):
+        # This is how it's implemented in the official SDK
+        rand = SystemRandom()
+        state = "".join(rand.choice(chars) for _ in range(length))
+        return state
+
+    def _get_redirect_uri(self):
+        domain = settings.BASE_BACKEND_URL
+        api_uri = self.API_URI
+        redirect_uri = f"{domain}{api_uri}"
+        return redirect_uri
+
+    def get_authorization_url(self):
+        redirect_uri = self._get_redirect_uri()
+
+        state = self._generate_state_session_token()
+
+        params = {
+            "response_type": "code",
+            "client_id": self._credentials.client_id,
+            "redirect_uri": redirect_uri,
+            "scope": " ".join(self.SCOPES),
+            "state": state,
+            "access_type": "offline",
+            "include_granted_scopes": "true",
+            "prompt": "select_account",
+        }
+
+        query_params = urlencode(params)
+        authorization_url = f"{self.GOOGLE_AUTH_URL}?{query_params}"
+
+        return authorization_url, state
+
+    def get_tokens(self, *, code: str) -> GoogleAccessTokens:
+        redirect_uri = self._get_redirect_uri()
+
+        # Reference: https://developers.google.com/identity/protocols/oauth2/web-server#obtainingaccesstokens
+        data = {
+            "code": code,
+            "client_id": self._credentials.client_id,
+            "client_secret": self._credentials.client_secret,
+            "redirect_uri": redirect_uri,
+            "grant_type": "authorization_code",
+        }
+
+        response = requests.post(self.GOOGLE_ACCESS_TOKEN_OBTAIN_URL, data=data)
+
+        if not response.ok:
+            raise ApplicationError("Failed to obtain access token from Google.")
+
+        tokens = response.json()
+        google_tokens = GoogleAccessTokens(id_token=tokens["id_token"], access_token=tokens["access_token"])
+
+        return google_tokens
+
+    def get_user_info(self, *, google_tokens: GoogleAccessTokens) -> Dict[str, Any]:
+        access_token = google_tokens.access_token
+        # Reference: https://developers.google.com/identity/protocols/oauth2/web-server#callinganapi
+        response = requests.get(self.GOOGLE_USER_INFO_URL, params={"access_token": access_token})
+
+        if not response.ok:
+            raise ApplicationError("Failed to obtain user info from Google.")
+
+        return response.json()
+
+
+def google_raw_login_get_credentials() -> GoogleRawLoginCredentials:
+    client_id = settings.GOOGLE_OAUTH2_CLIENT_ID
+    client_secret = settings.GOOGLE_OAUTH2_CLIENT_SECRET
+    project_id = settings.GOOGLE_OAUTH2_PROJECT_ID
+
+    if not client_id:
+        raise ImproperlyConfigured("GOOGLE_OAUTH2_CLIENT_ID missing in env.")
+
+    if not client_secret:
+        raise ImproperlyConfigured("GOOGLE_OAUTH2_CLIENT_SECRET missing in env.")
+
+    if not project_id:
+        raise ImproperlyConfigured("GOOGLE_OAUTH2_PROJECT_ID missing in env.")
+
+    credentials = GoogleRawLoginCredentials(client_id=client_id, client_secret=client_secret, project_id=project_id)
+
+    return credentials

styleguide_example/blog_examples/google_login_server_flow/raw/urls.py

@@ -0,0 +1,11 @@
+from django.urls import path
+
+from styleguide_example.blog_examples.google_login_server_flow.raw.apis import (
+    GoogleLoginApi,
+    GoogleLoginRedirectApi,
+)
+
+urlpatterns = [
+    path("callback/", GoogleLoginApi.as_view(), name="callback-raw"),
+    path("redirect/", GoogleLoginRedirectApi.as_view(), name="redirect-raw"),
+]