Fix password reset flow so forgot-password emails work (#1)

Replace the broken csrf_exempt handler (responses never returned) with a
working DRF endpoint, use FRONTEND_BASE_URL for reset links, and harden
set-password against reuse and short passwords.

llm_be/chat_backend/models.py

@@ -74,7 +74,10 @@ class CustomUser(AbstractUser):
     )
 
     def get_set_password_url(self):
-        return f"https://chat.aimloperations.com/set_password?slug={self.slug}"
+        from django.conf import settings
+
+        base = settings.FRONTEND_BASE_URL.rstrip("/")
+        return f"{base}/set_password/?slug={self.slug}"
 
 
 FEEDBACK_CHOICE = (

llm_be/chat_backend/tests/test_models.py

@@ -72,7 +72,7 @@ class CompanyAndUserTestCase(TestCase):
         user = make_user(email="person@example.com")
         self.assertEqual(
             user.get_set_password_url(),
-            f"https://chat.aimloperations.com/set_password?slug={user.slug}",
+            f"http://localhost:3000/set_password/?slug={user.slug}",
         )
 
     def test_user_defaults(self):

llm_be/chat_backend/tests/test_views_users.py

@@ -1,5 +1,6 @@
 from django.core import mail
 from django.urls import reverse
+from unittest import mock
 from rest_framework import status
 from rest_framework.test import APITestCase
 from rest_framework_simplejwt.tokens import RefreshToken
@@ -179,6 +180,8 @@ class SetPasswordTestCase(APITestCase):
         self.assertEqual(self.client.get(url).status_code, status.HTTP_200_OK)
 
     def test_post_sets_password(self):
+        self.user.set_unusable_password()
+        self.user.save()
         url = reverse("set_password", kwargs={"slug": self.user.slug})
 
         response = self.client.post(url, {"password": "brandnewpass"}, format="json")
@@ -187,6 +190,84 @@ class SetPasswordTestCase(APITestCase):
         self.user.refresh_from_db()
         self.assertTrue(self.user.check_password("brandnewpass"))
 
+    def test_post_rejects_user_that_already_has_a_password(self):
+        url = reverse("set_password", kwargs={"slug": self.user.slug})
+
+        response = self.client.post(url, {"password": "brandnewpass"}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
+
+    def test_post_rejects_short_password(self):
+        self.user.set_unusable_password()
+        self.user.save()
+        url = reverse("set_password", kwargs={"slug": self.user.slug})
+
+        response = self.client.post(url, {"password": "short"}, format="json")
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+
+class ResetPasswordTestCase(APITestCase):
+    def setUp(self):
+        self.user = make_user(email="person@example.com", password="testpass123")
+        self.url = reverse("reset_password")
+
+    def _mock_captcha(self, success=True, score=None):
+        result = {"success": success}
+        if score is not None:
+            result["score"] = score
+        return mock.patch(
+            "chat_backend.views.requests.post",
+            return_value=mock.Mock(json=mock.Mock(return_value=result)),
+        )
+
+    def test_reset_invalidates_password_and_sends_email(self):
+        with self._mock_captcha(success=True):
+            response = self.client.post(
+                self.url,
+                {"email": self.user.email, "recaptchaToken": "token"},
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.user.refresh_from_db()
+        self.assertFalse(self.user.has_usable_password())
+        self.assertEqual(len(mail.outbox), 1)
+        self.assertIn(self.user.slug, mail.outbox[0].body)
+        self.assertEqual(mail.outbox[0].to, [self.user.email])
+
+    def test_reset_unknown_email_still_returns_ok(self):
+        with self._mock_captcha(success=True):
+            response = self.client.post(
+                self.url,
+                {"email": "missing@example.com", "recaptchaToken": "token"},
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(len(mail.outbox), 0)
+        self.user.refresh_from_db()
+        self.assertTrue(self.user.has_usable_password())
+
+    def test_reset_rejects_failed_captcha(self):
+        with self._mock_captcha(success=False):
+            response = self.client.post(
+                self.url,
+                {"email": self.user.email, "recaptchaToken": "bad"},
+                format="json",
+            )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_reset_rejects_missing_email(self):
+        with self._mock_captcha(success=True):
+            response = self.client.post(
+                self.url, {"recaptchaToken": "token"}, format="json"
+            )
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
 
 class AcknowledgeTermsOfServiceTestCase(APITestCase):
     def test_post_marks_tos_signed(self):

llm_be/chat_backend/urls.py

@@ -21,7 +21,6 @@ from .views import (
     UserConversationAnalytics,
     CompanyUsageAnalytics,
     AdminAnalytics,
-    reset_password,
     DocumentWorkspaceView,
     DocumentUploadView,
     DocumentDetailView,
@@ -35,7 +34,9 @@ urlpatterns = [
     path("user/create/", CustomUserCreate.as_view(), name="create_user"),
     path("public/settings/", PublicSettingsView.as_view(), name="public_settings"),
     path("user/invite/", CustomUserInvite.as_view(), name="invite_user"),
-    path("user/reset_password/", reset_password, name="reset_password"),
+    path(
+        "user/reset_password/", ResetUserPassword.as_view(), name="reset_password"
+    ),
     path(
         "user/set_password/<slug:slug>/", SetUserPassword.as_view(), name="set_password"
     ),

llm_be/chat_backend/views.py

@@ -62,7 +62,7 @@ import pytz
 from langchain_ollama import OllamaEmbeddings
 
 from dateutil.relativedelta import relativedelta
-from django.views.decorators.csrf import csrf_exempt
+import requests
 
 from .utils import last_day_of_month
 from .services.llm_service import AsyncLLMService
@@ -154,18 +154,16 @@ def send_invite_email(slug, email_to_invite):
     msg.send(fail_silently=True)
 
 
-def send_password_reset_email(slug, email_to_invite):
-    logger.info("Sending reset email")
-    logger.info(f"url : https://www.chat.aimloperations.com/set_password?slug={slug}")
-    url = f"https://www.chat.aimloperations.com/set_password?slug={slug}"
-    subject = "Password reset for AI ML Operations, LLC Chat Services"
+def send_password_reset_email(slug, email_to):
+    logger.info("Sending Password reset email")
+    base = settings.FRONTEND_BASE_URL.rstrip("/")
+    url = f"{base}/set_password/?slug={slug}"
+    subject = "Password reset for Chat by AI ML Operations, LLC"
     from_email = "ryan@aimloperations.com"
-    to = email_to_invite
     d = {"url": url}
     html_content = get_template(r"emails/reset_email.html").render(d)
     text_content = get_template(r"emails/reset_email.txt").render(d)
-
-    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
+    msg = EmailMultiAlternatives(subject, text_content, from_email, [email_to])
     msg.attach_alternative(html_content, "text/html")
     msg.send(fail_silently=True)
 
@@ -184,20 +182,6 @@ def send_feedback_email(feedback_obj):
     msg.send(fail_silently=True)
 
 
-def send_password_reset_email(slug, email_to_invite):
-    logger.info("Sending Password reset email")
-    url = f"https://www.chat.aimloperations.com/set_password?slug={slug}"
-    subject = "Password reset for Chat by AI ML Operations, LLC"
-    from_email = "ryan@aimloperations.com"
-    to = email_to_invite
-    d = {"url": url}
-    html_content = get_template(r"emails/reset_email.html").render(d)
-    text_content = get_template(r"emails/reset_email.txt").render(d)
-    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
-    msg.attach_alternative(html_content, "text/html")
-    msg.send(fail_silently=True)
-
-
 class CustomUserInvite(APIView):
     http_method_names = ["post"]
 
@@ -234,66 +218,51 @@ class CustomUserInvite(APIView):
         return Response(status=status.HTTP_201_CREATED)
 
 
-@csrf_exempt
-def reset_password(request):
-    if request.method == "POST":
-        data = json.loads(request.body)
-        token = data.get("recaptchaToken")
-        payload = {
-            "secret": settings.CAPTCHA_SECRET_KEY,
-            "response": token,
-        }
-        response = requests.post(
-            "https://www.google.com/recaptcha/api/siteverify", data=payload
-        )
-        result = response.json()
-        if result.get("success") and result.get("score") >= 0.5:
-            email = data.get("email")
-            user = CustomUser.objects.filter(email=email).first()
-            if user:
-                user.set_unusable_password()
-                user.save()
-
-                # send the email
-                send_password_reset_email(user.slug, email)
-                JsonResponse(status=200)
-
-    JsonResponse(status=400)
-
-
 class ResetUserPassword(APIView):
-    http_method_names = [
-        "post",
-    ]
+    """Request a password-reset email. Invalidates the current password when sent."""
+
+    http_method_names = ["post"]
     permission_classes = (permissions.AllowAny,)
     authentication_classes = ()
 
     def post(self, request, format="json"):
-        """
-        Send an email with a set password link to the set password page
-        Also disable the account
-        """
-        logger.info(f"Password reset for requests. {request.data}")
+        logger.info("Password reset requested")
+        email = request.data.get("email")
         token = request.data.get("recaptchaToken")
+        if not email:
+            return Response(status=status.HTTP_400_BAD_REQUEST)
+
         payload = {
             "secret": settings.CAPTCHA_SECRET_KEY,
-            "response": recaptchaToken,
+            "response": token,
         }
-        response = requests.post(
-            "https://www.google.com/recaptcha/api/siteverify", data=payload
-        )
-        result = response.json()
-        if result.get("success") and result.get("score") >= 0.5:
-            user = CustomUser.objects.filter(email=email).first()
-            if user:
-                user.set_unusable_password()
-                user.save()
+        try:
+            captcha_response = requests.post(
+                "https://www.google.com/recaptcha/api/siteverify",
+                data=payload,
+                timeout=10,
+            )
+            result = captcha_response.json()
+        except requests.RequestException as exc:
+            logger.error("Captcha verification request failed: %s", exc)
+            return Response(status=status.HTTP_400_BAD_REQUEST)
 
-                # send the email
-                send_password_reset_email(user.slug, email)
-        else:
-            logger.error("Captcha secret failed")
+        # v2 invisible returns success only; v3 also returns a score.
+        if not result.get("success"):
+            logger.error("Captcha verification failed: %s", result)
+            return Response(status=status.HTTP_400_BAD_REQUEST)
+        score = result.get("score")
+        if score is not None and score < 0.5:
+            logger.error("Captcha score too low: %s", score)
+            return Response(status=status.HTTP_400_BAD_REQUEST)
 
+        user = CustomUser.objects.filter(email=email).first()
+        if user:
+            user.set_unusable_password()
+            user.save(update_fields=["password"])
+            send_password_reset_email(user.slug, email)
+
+        # Always 200 after valid captcha to avoid email enumeration.
         return Response(status=status.HTTP_200_OK)
 
 
@@ -303,15 +272,30 @@ class SetUserPassword(APIView):
     authentication_classes = ()
 
     def get(self, request, slug):
-        user = CustomUser.objects.get(slug=slug)
+        try:
+            user = CustomUser.objects.get(slug=slug)
+        except CustomUser.DoesNotExist:
+            return Response(status=status.HTTP_404_NOT_FOUND)
         if user.has_usable_password():
             return Response(status=status.HTTP_401_UNAUTHORIZED)
-        else:
-            return Response(status=status.HTTP_200_OK)
+        return Response(status=status.HTTP_200_OK)
 
     def post(self, request, slug, format="json"):
-        user = CustomUser.objects.get(slug=slug)
-        user.set_password(request.data["password"])
+        try:
+            user = CustomUser.objects.get(slug=slug)
+        except CustomUser.DoesNotExist:
+            return Response(status=status.HTTP_404_NOT_FOUND)
+        if user.has_usable_password():
+            return Response(status=status.HTTP_401_UNAUTHORIZED)
+
+        password = request.data.get("password")
+        if not password or len(password) < 8:
+            return Response(
+                {"password": "Password must be at least 8 characters."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        user.set_password(password)
         user.save()
         return Response(status=status.HTTP_200_OK)