Implement basic login with oauth

aleksis/core/settings.py

@@ -656,3 +656,22 @@ else:
 HAYSTACK_SEARCH_RESULTS_PER_PAGE = 10
 
 DJANGO_EASY_AUDIT_WATCH_REQUEST_EVENTS = False
+
+
+if _settings.get("auth.oauth2.enabled", False):
+    AUTHLIB_OAUTH_CLIENTS = {
+        "default": {
+            "client_id": _settings.get("auth.oauth2.id", ""),
+            "client_secret": _settings.get("auth.oauth2.secret", ""),
+            "request_token_url": _settings.get("auth.oauth2.token_url", ""),
+#            "request_token_params": _settings.get("auth.oauth2.token_params", None),
+            "access_token_url": _settings.get("auth.oauth2.access_url", ""),
+#            "access_token_params": _settings.get("auth.oauth2.access_params", None),
+#            "refresh_token_url": _settings.get("auth.oauth2.refresh)_token_url", None),
+            "authorize_url": _settings.get("auth.oauth2.authorize_url", ""),
+#            "api_base_url": _settings.get("auth.oauth2.api_url", ""),
+#            "client_kwargs": _settings.get("auth.oauth2.kwargs", None)
+        }
+    }
+
+    LOGIN_URL = "oauth_login"

aleksis/core/urls.py

@@ -158,6 +158,11 @@ if hasattr(settings, "TWILIO_ACCOUNT_SID"):
 
     urlpatterns += [path("", include(tf_twilio_urls))]
 
+# Add OAuth2 login view if enabled
+if hasattr(settings, "AUTHLIB_OAUTH_CLIENTS"):
+    urlpatterns += [path("oauth2/login", views.oauth_login, name="oauth_login")]
+    urlpatterns += [path("oauth2/authorize", views.oauth_authorize, name="oauth_authorize")]
+
 # Serve javascript-common if in development
 if settings.DEBUG:
     urlpatterns.append(path("__debug__/", include(debug_toolbar.urls)))

aleksis/core/util/oauth.py

+from django.dispatch import receiver
+from authlib.integrations.django_client import token_update
+  
+@receiver(token_update)
+def on_token_update(sender, token, refresh_token=None, access_token=None):
+    """Auto update oauth token."""
+    if refresh_token:
+        token = OAuth2Token.find(name=name, refresh_token=refresh_token)
+    elif access_token:
+        token = OAuth2Token.find(name=name, access_token=access_token)
+    else:
+        return
+  
+    # Update old token
+    token.access_token = token['access_token']
+    token.refresh_token = token.get('refresh_token')
+    token.expires_at = token['expires_at']
+    token.save()

aleksis/core/views.py

@@ -7,6 +7,7 @@ from django.core.exceptions import PermissionDenied
 from django.core.paginator import Paginator
 from django.http import HttpRequest, HttpResponse, HttpResponseNotFound
 from django.shortcuts import get_object_or_404, redirect, render
+from django.urls import reverse
 from django.utils.translation import gettext_lazy as _
 
 from django_tables2 import RequestConfig
@@ -584,3 +585,33 @@ def delete_group_type(request: HttpRequest, id_: int) -> HttpResponse:
     messages.success(request, _("The group type has been deleted."))
 
     return redirect("group_types")
+
+
+def oauth_login(request: HttpRequest) -> HttpResponse:
+    """Redirect to OAuth2 provider."""
+    # Build url and redirect
+    from authlib.integrations.django_client import OAuth  # noqa
+    oauth = OAuth()
+
+    oauth.register("default")
+    redirect_uri = request.build_absolute_uri(reverse("oauth_authorize"))
+    return oauth.default.authorize_redirect(request, redirect_uri)
+
+
+def oauth_authorize(request: HttpRequest) -> HttpResponse:
+    """Get token from oauth provider."""
+    from authlib.integrations.django_client import OAuth  # noqa
+
+    oauth = OAuth()
+
+    oauth.register("default")
+    token = oauth.default.authorize_access_token(request)
+
+    # Get email address from OAuth provider, find user and login
+    resp = oauth.default.get("user", token=token)
+    user = get_user_model().objects.get(email=resp["user"]["email"])
+    if user:
+        login(request, user)
+        return redirect("index")
+    else:
+        raise PermissionDenied("No user found for OAuth user.")