Add support for redirecting to /chatbot after login via ?next param (#1713)

* Add support for redirecting to /chatbot after login via ?next param

* Update redirect URL for console view test

The test now expects the redirect to include the 'next' parameter in the login URL, reflecting changes in authentication flow.

* Secure redirect for authenticated users in LoginView

LoginView now validates the 'next' parameter using Django's url_has_allowed_host_and_scheme to prevent open redirect vulnerabilities. Added a test to ensure unsafe 'next' URLs redirect to root instead of external sites.

Author: mabashian

ansible_ai_connect/main/base_views.py

@@ -58,7 +58,7 @@ def dispatch(self, request, *args, **kwargs):
             return handler(request, *args, **kwargs)
 
         except exceptions.NotAuthenticated:
-            return HttpResponseRedirect("/login")
+            return HttpResponseRedirect(f"/login?next={request.path}")
 
         except Exception as exc:
             # Map _internal_ errors to a generic PermissionDenied error

ansible_ai_connect/main/tests/test_console_views.py

@@ -38,7 +38,7 @@ def test_authentication_error(self, *args):
         # self.client.force_authenticate(user=self.user)
         r = self.client.get(reverse("console"))
         self.assertEqual(r.status_code, HTTPStatus.FOUND)
-        self.assertEqual(r.url, "/login")
+        self.assertEqual(r.url, "/login?next=/console/")
 
     @patch.object(IsOrganisationAdministrator, "has_permission", return_value=True)
     @patch.object(IsOrganisationLightspeedSubscriber, "has_permission", return_value=True)

ansible_ai_connect/main/tests/test_views.py

@@ -130,6 +130,26 @@ class MockUser:
         self.assertTrue(isinstance(response, HttpResponseRedirect))
         self.assertEqual(response.url, "/")
 
+    def test_no_login_for_auth_user_with_next(self):
+        class MockUser:
+            is_authenticated = True
+
+        request = RequestFactory().get("/login?next=/chatbot", SERVER_NAME="testserver")
+        request.user = MockUser()
+        response = LoginView.as_view()(request)
+        self.assertTrue(isinstance(response, HttpResponseRedirect))
+        self.assertEqual(response.url, "/chatbot")
+
+    def test_no_login_for_auth_user_with_unsafe_next(self):
+        class MockUser:
+            is_authenticated = True
+
+        request = RequestFactory().get("/login?next=http://malicious-site.com")
+        request.user = MockUser()
+        response = LoginView.as_view()(request)
+        self.assertTrue(isinstance(response, HttpResponseRedirect))
+        self.assertEqual(response.url, "/")
+
 
 @override_settings(ALLOW_METRICS_FOR_ANONYMOUS_USERS=False)
 class TestMetricsView(APITransactionTestCase):
@@ -343,7 +363,7 @@ def test_chatbot_link_with_rh_user(self):
     def test_chatbot_view_with_anonymous_user(self):
         r = self.client.get(reverse("chatbot"))
         self.assertEqual(r.status_code, HTTPStatus.FOUND)
-        self.assertEqual(r.url, "/login")
+        self.assertEqual(r.url, "/login?next=/chatbot/")
 
     def test_chatbot_view_with_non_rh_user(self):
         self.client.force_login(user=self.non_rh_user)

ansible_ai_connect/main/views.py

@@ -21,6 +21,7 @@
 from django.contrib.auth import views as auth_views
 from django.contrib.auth.models import AnonymousUser
 from django.http import HttpResponseRedirect
+from django.utils.http import url_has_allowed_host_and_scheme
 from django_prometheus.exports import ExportToDjangoView
 from oauth2_provider.contrib.rest_framework import IsAuthenticatedOrTokenHasScope
 from rest_framework.exceptions import PermissionDenied
@@ -50,6 +51,7 @@
 class LoginView(auth_views.LoginView):
     def get_context_data(self, **kwargs):
         context = super().get_context_data(**kwargs)
+        context["next"] = self.request.GET.get("next")
         context["deployment_mode"] = settings.DEPLOYMENT_MODE
         context["project_name"] = settings.ANSIBLE_AI_PROJECT_NAME
         context["aap_api_provider_name"] = settings.AAP_API_PROVIDER_NAME
@@ -58,6 +60,11 @@ def get_context_data(self, **kwargs):
 
     def dispatch(self, request, *args, **kwargs):
         if self.request.user.is_authenticated:
+            next_url = self.request.GET.get("next", "/")
+            if url_has_allowed_host_and_scheme(
+                next_url, allowed_hosts={request.get_host()}, require_https=request.is_secure()
+            ):
+                return HttpResponseRedirect(next_url)
             return HttpResponseRedirect("/")
         return super().dispatch(request, *args, **kwargs)