Notify users on login via e-mail

docker/prod/.env.example

@@ -4,7 +4,7 @@ SECRET_KEY="changeme"
 PRIMARY_HOST="https://test.hypha.app"
 EMAIL_HOST="hypha.app"
 
-EMAIL_SUBJECT_PREFIX="[Hypha]"
+EMAIL_SUBJECT_PREFIX="[Hypha] "
 ORG_EMAIL="hello@hypha.app"
 SERVER_EMAIL="test@hypha.app"
 

hypha/apply/users/apps.py

@@ -0,0 +1,8 @@
+from django.apps import AppConfig
+
+
+class UsersConfig(AppConfig):
+    name = "hypha.apply.users"
+
+    def ready(self):
+        from . import signals  # NOQA

hypha/apply/users/signals.py

@@ -0,0 +1,57 @@
+from django.conf import settings
+from django.contrib.auth.signals import user_logged_in
+from django.dispatch import receiver
+from django.utils import formats, timezone
+from django.utils.translation import gettext_lazy as _
+from wagtail.models import Site
+
+from hypha.core.mail import MarkdownMail
+
+from .utils import get_zoneinfo
+
+HIJACK_VIEW_NAMES = {
+    "hijack-become",
+    "users:hijack",
+    "hijack:acquire",
+    "hijack:release",
+}
+
+
+@receiver(user_logged_in)
+def send_login_notification(sender, request, user, **kwargs):
+    if not settings.SEND_MESSAGES or not user.email:
+        return
+
+    if getattr(user, "backend", "").startswith("social_core."):
+        return
+
+    if request and getattr(request, "resolver_match", None):
+        if request.resolver_match.view_name in HIJACK_VIEW_NAMES:
+            return
+
+    tz_name = (
+        getattr(request, "session", {}).get("user_timezone", "") if request else ""
+    )
+    user_tz = get_zoneinfo(tz_name)
+
+    subject = _("Successful login to %(org)s") % {"org": settings.ORG_LONG_NAME}
+    if settings.EMAIL_SUBJECT_PREFIX:
+        subject = str(settings.EMAIL_SUBJECT_PREFIX) + str(subject)
+
+    email = MarkdownMail("users/emails/login_notification.md")
+    email.send(
+        to=user.email,
+        subject=subject,
+        from_email=settings.DEFAULT_FROM_EMAIL,
+        context={
+            "user": user,
+            "login_time": "{} ({})".format(
+                formats.date_format(
+                    timezone.localtime(timezone=user_tz), "SHORT_DATETIME_FORMAT"
+                ),
+                tz_name or timezone.get_current_timezone_name(),
+            ),
+            "site": Site.find_for_request(request) if request else None,
+            "ORG_EMAIL": settings.ORG_EMAIL,
+        },
+    )

hypha/apply/users/templates/users/emails/login_notification.md

@@ -0,0 +1,19 @@
+{% load i18n wagtailadmin_tags %}{% base_url_setting as base_url %}
+{% blocktrans %}Dear {{ user }},{% endblocktrans %}
+
+{% blocktrans %}This is to notify you that your account was successfully logged in to {{ ORG_LONG_NAME }}.{% endblocktrans %}
+
+{% blocktrans with login_time=login_time %}Login time: {{ login_time }}{% endblocktrans %}
+
+{% blocktrans %}If you did not log in, please contact us immediately and consider changing your password.{% endblocktrans %}
+
+{% if ORG_EMAIL %}
+{% blocktrans %}If you have any questions, please contact us at {{ ORG_EMAIL }}.{% endblocktrans %}
+{% endif %}
+
+{% blocktrans %}Kind Regards,
+The {{ ORG_SHORT_NAME }} Team{% endblocktrans %}
+
+--
+{{ ORG_LONG_NAME }}
+{% if site %}{{ site.root_url }}{% else %}{{ base_url }}{% endif %}

hypha/apply/users/tests/test_signals.py

@@ -0,0 +1,80 @@
+from unittest.mock import MagicMock
+
+from django.contrib.auth.signals import user_logged_in
+from django.core import mail
+from django.test import RequestFactory, TestCase, override_settings
+
+from .factories import UserFactory
+
+
+@override_settings(SEND_MESSAGES=True)
+class TestSendLoginNotification(TestCase):
+    def setUp(self):
+        self.factory = RequestFactory()
+        self.user = UserFactory()
+
+    def _fire_signal(self, user=None, request=None):
+        if user is None:
+            user = self.user
+        if request is None:
+            request = self.factory.get("/")
+        user_logged_in.send(sender=user.__class__, request=request, user=user)
+
+    def test_sends_email_on_login(self):
+        self._fire_signal()
+        self.assertEqual(len(mail.outbox), 1)
+
+    def test_email_sent_to_user(self):
+        self._fire_signal()
+        self.assertIn(self.user.email, mail.outbox[0].to)
+
+    def test_email_subject_contains_org_name(self):
+        from django.conf import settings
+
+        self._fire_signal()
+        self.assertIn(settings.ORG_LONG_NAME, mail.outbox[0].subject)
+
+    def test_no_email_when_send_messages_disabled(self):
+        with self.settings(SEND_MESSAGES=False):
+            self._fire_signal()
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_no_email_when_user_has_no_email(self):
+        self.user.email = ""
+        self.user.save()
+        self._fire_signal()
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_no_email_when_request_is_none(self):
+        # Signal can be fired without a request (e.g. management commands)
+        self._fire_signal(request=None)
+        self.assertEqual(len(mail.outbox), 1)
+
+    def test_email_body_contains_login_time(self):
+        self._fire_signal()
+        self.assertTrue(any("Login time" in part for part in [mail.outbox[0].body]))
+
+    def _fire_signal_with_view_name(self, view_name):
+        request = self.factory.get("/")
+        request.resolver_match = MagicMock(view_name=view_name)
+        self._fire_signal(request=request)
+
+    def test_no_email_on_hijack_acquire(self):
+        self._fire_signal_with_view_name("hijack:acquire")
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_no_email_on_hijack_release(self):
+        self._fire_signal_with_view_name("hijack:release")
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_no_email_on_hijack_become(self):
+        self._fire_signal_with_view_name("hijack-become")
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_no_email_on_users_hijack_view(self):
+        self._fire_signal_with_view_name("users:hijack")
+        self.assertEqual(len(mail.outbox), 0)
+
+    def test_email_sent_for_non_hijack_view(self):
+        self._fire_signal_with_view_name("account_login")
+        self.assertEqual(len(mail.outbox), 1)

hypha/apply/users/urls.py

@@ -37,6 +37,7 @@
     oauth,
     send_confirm_access_email_view,
     set_password_view,
+    set_timezone_view,
 )
 
 app_name = "users"
@@ -159,6 +160,7 @@
     ),
     path("activate/", create_password, name="activate_password"),
     path("oauth", oauth, name="oauth"),
+    path("set-timezone/", set_timezone_view, name="set_timezone"),
     # 2FA
     path("two_factor/setup/", TWOFASetupView.as_view(), name="setup"),
     path(

hypha/apply/users/utils.py

@@ -1,4 +1,5 @@
 import string
+import zoneinfo
 
 import nh3
 from django.conf import settings
@@ -196,6 +197,16 @@ def update_is_staff(request, user):
         user.save()
 
 
+def get_zoneinfo(tz_name):
+    """Return a ZoneInfo for tz_name, or None if invalid/empty."""
+    if not tz_name:
+        return None
+    try:
+        return zoneinfo.ZoneInfo(tz_name)
+    except (zoneinfo.ZoneInfoNotFoundError, KeyError):
+        return None
+
+
 def strip_html_and_nerf_urls(value: str):
     # Remove all HTML tags. This prohibits HTML without creating hurdles.
     cleaned_value = nh3.clean(value, tags=set())

hypha/apply/users/views.py

@@ -31,6 +31,7 @@
 from django.views.decorators.cache import never_cache
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.debug import sensitive_post_parameters
+from django.views.decorators.http import require_POST
 from django.views.generic import UpdateView
 from django.views.generic.base import TemplateView
 from django.views.generic.edit import FormView
@@ -70,6 +71,7 @@
 from .utils import (
     generate_numeric_token,
     get_redirect_url,
+    get_zoneinfo,
     send_activation_email,
     send_confirmation_email,
 )
@@ -870,6 +872,15 @@ def set_password_view(request):
         return HttpResponse(_("✓ Check your email for password set link."))
 
 
+@require_POST
+def set_timezone_view(request: HttpRequest) -> HttpResponse:
+    """Store the browser timezone in the session for use in login notifications."""
+    tz_name = request.POST.get("user_timezone", "")
+    if get_zoneinfo(tz_name):
+        request.session["user_timezone"] = tz_name
+    return HttpResponse(status=204)
+
+
 @never_cache
 @csrf_exempt
 @psa(f"{settings.SOCIAL_AUTH_URL_NAMESPACE}:complete")

hypha/settings/django.py

@@ -15,7 +15,7 @@
     "hypha.apply.dashboard",
     "hypha.apply.flags",
     "hypha.home",
-    "hypha.apply.users",
+    "hypha.apply.users.apps.UsersConfig",
     "hypha.apply.review",
     "hypha.apply.determinations",
     "hypha.apply.stream_forms",

hypha/templates/base.html

@@ -202,6 +202,15 @@
                 });
             </script>
         {% endif %}
+        {% if not request.session.user_timezone %}
+            <script>
+                fetch("{% url 'users:set_timezone' %}", {
+                    method: 'POST',
+                    headers: {'Content-Type': 'application/x-www-form-urlencoded', 'X-CSRFToken': '{{ csrf_token }}'},
+                    body: 'user_timezone=' + encodeURIComponent(Intl.DateTimeFormat().resolvedOptions().timeZone)
+                });
+            </script>
+        {% endif %}
         {% include "includes/body_end.html" %}
     </body>
 </html>