feat: Sso backend 34 (#166)

* add google auth backend

* fix

* cache access token and create user

* improve caching

* update migrations

* fix create user

* BaseOAuth2LoginForm

* cache token object

* create cache folder

* add encrypted char field

* plural

* plural

* auto increase max length

* new common

* split user proxies into separate files

* fix

* fix linting errors

* fix

* fix type aliases

* fix linting errors

Author: SKairinos

Pipfile

@@ -28,11 +28,11 @@ pyjwt = "==2.6.0" # TODO: upgrade to latest version.
 psutil = "==7.0.0"
 importlib-metadata = "==4.13.0" # TODO: remove. needed by old portal
 django-formtools = "==2.5.1" # TODO: remove. needed by old portal
-django-otp = "==1.6.0" # TODO: remove. needed by old portal
+django-otp = "==1.6.1" # TODO: remove. needed by old portal
 # https://pypi.org/user/codeforlife/
-cfl-common = "==8.7.9" # TODO: remove
-codeforlife-portal = "==8.7.9" # TODO: remove
-rapid-router = "==7.5.17" # TODO: remove
+cfl-common = "==8.8.2" # TODO: remove
+codeforlife-portal = "==8.8.2" # TODO: remove
+rapid-router = "==7.5.21" # TODO: remove
 phonenumbers = "==8.12.12" # TODO: remove
 
 [dev-packages]

Pipfile.lock

@@ -0,0 +1,124 @@
+"""
+© Ocado Group
+Created on 11/08/2025 at 11:07:45(+01:00).
+"""
+
+import typing as t
+
+from django.core.cache import cache
+
+K = t.TypeVar("K")
+V = t.TypeVar("V")
+
+
+class BaseValueCache(t.Generic[V]):
+    """Base class which helps to get and set cache values."""
+
+    @classmethod
+    def get(
+        cls,
+        key: str,
+        default: t.Optional[V] = None,
+        version: t.Optional[int] = None,
+    ) -> t.Optional[V]:
+        """
+        Fetch a given key from the cache. If the key does not exist, return
+        default, which itself defaults to None.
+        """
+        return cache.get(
+            key=key,
+            default=default,
+            version=version,
+        )
+
+    @classmethod
+    def set(
+        cls,
+        key: str,
+        value: V,
+        timeout: t.Optional[int] = None,
+        version: t.Optional[int] = None,
+    ):
+        """
+        Set a value in the cache. If timeout is given, use that timeout for the
+        key; otherwise use the default cache timeout.
+        """
+        cache.set(
+            key=key,
+            value=value,
+            timeout=timeout,
+            version=version,
+        )
+
+
+class BaseFixedKeyValueCache(BaseValueCache[V], t.Generic[V]):
+    """Base class which helps to get and set cache values with a fixed key."""
+
+    key: str
+
+    # pylint: disable=arguments-differ
+
+    @classmethod
+    def get(  # type: ignore[override]
+        cls,
+        default: t.Optional[V] = None,
+        version: t.Optional[int] = None,
+    ) -> t.Optional[V]:
+        return super().get(
+            key=cls.key,
+            default=default,
+            version=version,
+        )
+
+    @classmethod
+    def set(  # type: ignore[override]
+        cls,
+        value: V,
+        timeout: t.Optional[int] = None,
+        version: t.Optional[int] = None,
+    ):
+        super().set(
+            key=cls.key,
+            value=value,
+            timeout=timeout,
+            version=version,
+        )
+
+    # pylint: enable=arguments-differ
+
+
+class BaseDynamicKeyValueCache(BaseValueCache[V], t.Generic[K, V]):
+    """Base class which helps to get and set cache values with a dynamic key."""
+
+    @staticmethod
+    def make_key(key: K) -> str:
+        """Make the cache key from the key's data."""
+        raise NotImplementedError()
+
+    @classmethod
+    def get(  # type: ignore[override]
+        cls,
+        key: K,
+        default: t.Optional[V] = None,
+        version: t.Optional[int] = None,
+    ) -> t.Optional[V]:
+        return super().get(
+            key=cls.make_key(key),
+            default=default,
+            version=version,
+        )
+
+    @classmethod
+    def set(  # type: ignore[override]
+        cls,
+        key: K,
+        value: V,
+        timeout: t.Optional[int] = None,
+        version: t.Optional[int] = None,
+    ):
+        super().set(
+            key=cls.make_key(key),
+            value=value,
+            timeout=timeout,
+            version=version,
+        )

codeforlife/caches.py

@@ -79,3 +79,19 @@ def get_invalid_login_error_message(self) -> str:
             NotImplementedError: If message is not set.
         """
         raise NotImplementedError()
+
+
+class BaseOAuth2LoginForm(
+    BaseLoginForm[AnyAbstractBaseUser],
+    t.Generic[AnyAbstractBaseUser],
+):
+    """
+    Base login form that all other login forms using OAuth2.0 must inherit.
+    """
+
+    code = forms.CharField(min_length=1)
+    code_verifier = forms.CharField(min_length=43, max_length=128)
+    redirect_uri = forms.CharField(min_length=1)
+
+    def get_invalid_login_error_message(self):
+        return "Failed to exchange code for access token."

codeforlife/forms.py

@@ -7,3 +7,4 @@
 from .abstract_base_user import AbstractBaseUser
 from .base import *
 from .base_session_store import BaseSessionStore
+from .encrypted_char_field import EncryptedCharField

codeforlife/models/__init__.py

@@ -0,0 +1,72 @@
+"""
+© Ocado Group
+Created on 12/08/2025 at 10:28:24(+01:00).
+"""
+
+import typing as t
+
+from cryptography.fernet import Fernet
+from django.conf import settings
+from django.db import models
+
+
+class EncryptedCharField(models.CharField):
+    """
+    A custom CharField that encrypts data before saving and decrypts it when
+    retrieved.
+    """
+
+    _fernet = Fernet(settings.SECRET_KEY)
+    _prefix = "ENC:"
+
+    def __init__(self, *args, **kwargs):
+        kwargs["max_length"] += len(self._prefix)
+        super().__init__(*args, **kwargs)
+
+    # pylint: disable-next=unused-argument
+    def from_db_value(self, value: t.Optional[str], expression, connection):
+        """
+        Converts a value as returned by the database to a Python object. It is
+        the reverse of get_prep_value().
+
+        https://docs.djangoproject.com/en/5.1/howto/custom-model-fields/#converting-values-to-python-objects
+        """
+        if isinstance(value, str):
+            return self.decrypt_value(value)
+        return value
+
+    def to_python(self, value: t.Optional[str]):
+        """
+        Converts the value into the correct Python object. It acts as the
+        reverse of value_to_string(), and is also called in clean().
+
+        https://docs.djangoproject.com/en/5.1/howto/custom-model-fields/#converting-values-to-python-objects
+        """
+        if isinstance(value, str):
+            return self.decrypt_value(value)
+        return value
+
+    def get_prep_value(self, value: t.Optional[str]):
+        """
+        'value' is the current value of the model's attribute, and the method
+        should return data in a format that has been prepared for use as a
+        parameter in a query.
+
+        https://docs.djangoproject.com/en/5.1/howto/custom-model-fields/#converting-python-objects-to-query-values
+        """
+        if isinstance(value, str):
+            return self.encrypt_value(value)
+        return value
+
+    def encrypt_value(self, value: str):
+        """Encrypt the value if it's not encrypted."""
+        if not value.startswith(self._prefix):
+            return self._prefix + self._fernet.encrypt(value.encode()).decode()
+        return value
+
+    def decrypt_value(self, value: str):
+        """Decrpyt the value if it's encrypted.."""
+        if value.startswith(self._prefix):
+            value = value[len(self._prefix) :]
+            return self._fernet.decrypt(value).decode()
+        return value

codeforlife/models/encrypted_char_field.py

@@ -126,3 +126,11 @@ def get_redis_url():
 
 # The URL to connect to the Redis cache.
 REDIS_URL = get_redis_url()
+
+# Our Google OAuth 2.0 client credentials
+# https://console.cloud.google.com/auth/clients
+GOOGLE_CLIENT_ID = os.getenv(
+    "GOOGLE_CLIENT_ID",
+    "354656325390-o5n12nbaivhi4do8lalkh29q403uu9u4.apps.googleusercontent.com",
+)
+GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET", "REPLACE_ME")

codeforlife/settings/custom.py

@@ -149,6 +149,7 @@ def get_databases():
     "codeforlife.user.auth.backends.OtpBypassTokenBackend",
     "codeforlife.user.auth.backends.StudentBackend",
     "codeforlife.user.auth.backends.StudentAutoBackend",
+    "codeforlife.user.auth.backends.GoogleBackend",
 ]
 
 # Sessions

codeforlife/settings/django.py

@@ -47,3 +47,24 @@ def get_arg(cls: t.Type[t.Any], index: int, orig_base: int = 0):
         The type arg from the class.
     """
     return t.get_args(cls.__orig_bases__[orig_base])[index]
+
+
+class OAuth2TokenFromRefreshDict(t.TypedDict):
+    """An OAuth 2.0 token given in exchange for a refresh token."""
+
+    access_token: str
+    token_type: str
+    scope: str
+    expires_in: int
+    error: t.NotRequired[dict]
+
+
+class OAuth2TokenFromCodeDict(t.TypedDict):
+    """An OAuth 2.0 token given in exchange for a code."""
+
+    access_token: str
+    token_type: str
+    scope: str
+    expires_in: int
+    refresh_token: str
+    error: t.NotRequired[dict]

codeforlife/types.py

@@ -5,6 +5,7 @@
 
 # TODO: Create a custom auth backend for Django admin permissions
 from .email import EmailBackend
+from .google import GoogleBackend
 from .otp import OtpBackend
 from .otp_bypass_token import OtpBypassTokenBackend
 from .student import StudentBackend